Spark 1.3.1 - Паркет - Обнаружение нескольких разделов при сохранении

У меня есть приложение ETL, использующее Spark 1.3.1, Amazon S3 и EMR 3.8. Мне нужно мое приложение для сохранения Dataframe в нескольких разделах.

Согласно документации Spark 1.3.1, это способ разбиения вашего файла паркета:

Обнаружение разделов

df1.save("data/test_table/key=1", "parquet")

Но проблема в том, что у моего RDD много данных и их нужно сохранять на отдельные разделы. (в группе)

Я не могу найти правильную логику, чтобы эта программа работала быстро (или не так медленно)

Некоторые попытки:

#Get All RDD Partition's 
AllPartitions = RDD.map(lambda x: x[0]).distinct()

#For all partitions save filtering 
for part in AllPartitions.collect():
   filteredDF = df.filter(df.recordOpeningDate == part)
   df.save(path=outputFilePath + "/FIELD=" + part, source='parquet',mode='append')

Попытка использовать GroupBy

def Mapping(line):
   return (
   line[0] , [ 
   #Definition of my RDD
    ])

def SavePartitions(KV):
   #WRONG ! I cant create a DF inside a transformation 
   df = sqlContext.createDataFrame(KV, SCHEMA)
   df.save(...)    

RDD = RDD.map(Mapping).groupByKey().mapValues(SavePartitions)

python amazon-s3 parquet apache-spark emr
person Denys Nunes    schedule 02.09.2015    source источник

Ответы (2)


arrow_upward
0
arrow_downward

Возможно, я неправильно понимаю ваш вопрос, но если у вас есть dataframe, вы можете вызвать repartition, чтобы разделить ваши данные на нужное количество разделов, скажем, 10:

myDataFrame.repartition(10).saveAsParquet("/path/to/output/")

Это приведет к тому, что ваши данные будут храниться в 10 разных файлах в папке /path/to/output/.

person Glennie Helles Sindholt    schedule 04.09.2015
comment
Мое решение состояло в том, чтобы обновить мои кластеры для EMR 4.0 (с Spark 1.4.1) и использовать метод pyspark.sql.DataFrameWriter.save. - person Denys Nunes; 11.09.2015

arrow_upward
0
arrow_downward

Для информации,

Мое решение состояло в том, чтобы обновить мои кластеры для EMR 4.0 (с Spark 1.4.1) и использовать метод pyspark.sql.DataFrameWriter.save.

person Denys Nunes    schedule 11.09.2015