Записать DataFrame в паркет на HDFS, разделенном на несколько столбцов с помощью dynamici c partitionOverwriteMode - PullRequest
Новая
       4

Записать DataFrame в паркет на HDFS, разделенном на несколько столбцов с помощью dynamici c partitionOverwriteMode

0 голосов
/ 15 января 2020

У меня есть датафрейм, который я хочу сохранить в формате паркета в HDFS. Я хотел бы разделить его на несколько столбцов.

Когда я записываю данные в HDFS - сам каталог, и в нем создается только файл _SUCCESS, но данных нет. Я использую partitionOverwriteMode=dynamic и overwrite в качестве режима сохранения. К тому времени я исполняю код пути не существует. Если я изменю режим сохранения на append, то он работает нормально. Я также пытался писать в локальную файловую систему. В этом случае оба режима работают правильно. Если указан только 1 столбец раздела, то он тоже работает нормально.

Есть идеи, как сделать так, чтобы перезаписать , работает с разбиением на несколько столбцов? Любые советы приветствуются. Спасибо!

Пример кода: 

from pyspark.sql import SparkSession
from pyspark.conf import SparkConf

data = [
    {'country': 'DE', 'fk_imported_at': '20191212', 'user_id': 15},
    {'country': 'DE', 'fk_imported_at': '20191212', 'user_id': 14},
    {'country': 'US', 'fk_imported_at': '20191212', 'user_id': 12},
    {'country': 'US', 'fk_imported_at': '20191212', 'user_id': 13},
    {'country': 'DE', 'fk_imported_at': '20191213', 'user_id': 4},
    {'country': 'DE', 'fk_imported_at': '20191213', 'user_id': 2},
    {'country': 'US', 'fk_imported_at': '20191213', 'user_id': 1},
]


if __name__ == '__main__':
    conf = SparkConf()
    conf.set('spark.sql.sources.partitionOverwriteMode', 'dynamic')

    spark = (
        SparkSession
        .builder
        .config(conf=conf)
        .appName('test partitioning')
        .enableHiveSupport()
        .getOrCreate()
    )

    df = spark.createDataFrame(data)
    df.show()
    df.repartition(1).write.parquet('/tmp/spark_save_mode', 'overwrite', ['fk_imported_at', 'country'])

    spark.stop()

Я подаю заявку в режиме клиента. Версия Spark 2.3.0. * oop версия 2.6.0

...