Spark 1.6: Хранить фрейм данных в нескольких CSV-файлах в формате hdf (раздел по идентификатору) - PullRequest
Новая
       41

Spark 1.6: Хранить фрейм данных в нескольких CSV-файлах в формате hdf (раздел по идентификатору)

0 голосов
/ 25 марта 2020

Я пытаюсь сохранить dataFrame в раздел csv по id, для этого я использую spark 1.6 и scala. Функция partitionBy ("id") не дает мне правильный результат.

Мой код здесь: 

validDf.write
       .partitionBy("id")
       .format("com.databricks.spark.csv")
       .option("header", "true")
       .option("delimiter", ";")
       .mode("overwrite")       
       .save("path_hdfs_csv")

My Dataframe looks like  :
-----------------------------------------
| ID        |  NAME       |  STATUS     |
-----------------------------------------
|     1     |     N1      |     S1      |
|     2     |     N2      |     S2      |
|     3     |     N3      |     S1      |
|     4     |     N4      |     S3      |
|     5     |     N5      |     S2      |
-----------------------------------------

Этот код создает 3 CSV-раздела по умолчанию (part_0, part_1, part_2), не основанных на идентификаторе столбца.

Что я ожидаю, так это : получение sub dir или раздела для каждого идентификатора. Любая помощь?

1 Ответ

0 голосов
/ 26 марта 2020

Spark-csv в spark1.6 (или во всех версиях spark ниже 2) не поддерживает разбиение.
Ваш код будет работать для spark> 2.0.0.

Для вашей версии spark вам необходимо сначала подготовить csv и сохранить его как текст (разбиение работает для spark-text): 

import org.apache.spark.sql.functions.{col,concat_ws}
val key = col("ID")
val concat_col = concat_ws(",",df.columns.map(c=>col(c)):_*) // concat cols to one col
val final_df = df.select(col("ID"),concat_col) // dataframe with 2 columns: id and string 
final_df.write.partitionBy("ID").text("path_hdfs_csv") //save to hdfs
...