Збереження TDigest у зовнішню таблицю Hive з DataFrame за допомогою Spark3

3 Березня 2024 в 14:21 27

При роботі з Apache Spark3 і Hive можуть виникати труднощі зі збереженням даних типу TDigest у зовнішню таблицю Hive. TDigest, сам по собі, представляє собою тип користувацького визначення даних (UDT) Apache Spark і зазвичай серіалізується у форматі Parquet. Однак спроба зберегти його безпосередньо у зовнішню таблицю Hive може призвести до проблем сумісності та викликати помилки.

Конкретна помилка, з якою ви можете стикнутися, може виглядати так:

Cannot write 'td1': struct&lt;compression:double,maxDiscrete:int,cent:array&lt;double&gt;,mass:array&lt;double&gt;&gt; is incompatible with struct&lt;compression:double,maxDiscrete:int,cent:array&lt;double&gt;,mass:array&lt;double&gt;&gt;

1	Cannot write 'td1': struct<compression:double,maxDiscrete:int,cent:array<double>,mass:array<double>> is incompatible with struct<compression:double,maxDiscrete:int,cent:array<double>,mass:array<double>>

Ця помилка виникає через неправильне трактування типу TDigest класом DataType Spark, що може призводити до проблем сумісності при спробі вставки даних у таблицю Hive.

Для вирішення цієї проблеми одним із шляхів є збереження DataFrame, що містить тип TDigest, у вигляді файлів Parquet безпосередньо у розташування таблиці. Хоча цей підхід може здатися не найоптимальнішим, він ефективно обходить проблеми сумісності і забезпечує успішне збереження ваших даних.

Ось приклад Scala-коду, що демонструє, як це можна зробити:

val data = spark.createDataFrame(Vector.fill(1000) {
  (nextInt(10), nextGaussian)
})
val udf1 = TDigestAggregator.udf[Double](compression = 0.2, maxDiscrete = 25)
val agg = data.agg(udf1($"_1").as("td1"), udf1($"_2").as("td2"))
agg.write.parquet("шлях_до_розташування_таблиці")

val data = spark.createDataFrame(Vector.fill(1000) {

(nextInt(10), nextGaussian)

})

val udf1 = TDigestAggregator.udf[Double](compression = 0.2, maxDiscrete = 25)

val agg = data.agg(udf1($"_1").as("td1"), udf1($"_2").as("td2"))

agg.write.parquet("шлях_до_розташування_таблиці")

У вищезазначеному коді ми створюємо DataFrame ‘data’ з симульованими даними, визначаємо функцію користувача (UDF) ‘udf1’ для агрегування значень TDigest і обчислюємо результат агрегації ‘agg’. Нарешті, ми використовуємо метод ‘write.parquet’ для збереження агрегованого DataFrame безпосередньо у вказане розташування таблиці.

Використання цього обхідного рішення дозволить вам ефективно зберігати дані типу TDigest з DataFrame Apache Spark у зовнішню таблицю Hive, уникнувши проблем сумісності та забезпечивши безперешкодну інтеграцію та обробку даних.