Эффективный способ добавления Spark DataFrames в al oop с использованием pyspark - PullRequest
Новая
       28

Эффективный способ добавления Spark DataFrames в al oop с использованием pyspark

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

у меня есть '|' огромные текстовые файлы с разделителями, я хочу объединить все текстовые файлы и создать один огромный фрейм данных spark, он будет позже использован для процесса ETL с использованием pyspark.

неэффективный способ

1) Создайте пустой фрейм данных искры, df

2) В al oop прочитайте текстовый файл как спаркфрейм данных df1 и добавьте его в пустой фрейм данных искры df 

df = spark.createDataFrame([],schema)
for x in os.listdir(textfiles_dir):
    filepath = '{}/{}'.format(textfiles_dir,x)
    df1 = spark.read.format("csv") \
                    .option("header", "true") \
                    .option("delimiter", "|") \
                    .option("inferSchema","true") \
                    .load(filepath)
    df = df.union(df1)

Это не эффективный искровой способ.

Может ли кто-нибудь предложить эффективный способ сделать это? Было бы замечательно, если бы вы объяснили его с помощью примера кода.

Спасибо :)

Ответы [ 3 ]

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

Как уже отмечали другие, вы захотите читать весь каталог текстовых файлов как кадр данных, а не итеративно читать каждый отдельный каталог: 

df = spark.read.format("csv") \
                    .option("header", "true") \
                    .option("delimiter", "|") \
                    .option("inferSchema","true") \
                    .load(textfiles_dir)

Если вы действительно хотите go объединение маршрут, я бы рекомендовал использовать функцию union в SparkContext (http://spark.apache.org/docs/latest/api/python/pyspark.html?highlight=union#pyspark .SparkContext.union ) вместо функции объединения в DataFrame: 

dfs = []
for x in os.listdir(textfiles_dir):
   filepath = '{}/{}'.format(textfiles_dir,x)
   df1 = spark.read.format("csv") \
                .option("header", "true") \
                .option("delimiter", "|") \
                .option("inferSchema","true") \
                .load(filepath)
   dfs.append(df1)
df = spark.sparkContext.union(dfs)
0 голосов
/ 31 марта 2020

filepath = filepath каталога, в котором существует несколько файлов

dataframe = spark.read.format ("csv"). Option ("header", "true"). Option ("delimiter", "| ") .load (filepath)

0 голосов
/ 27 марта 2020
  1. df1 = spark.read. ... .load ("pathFolder /") - прочитать все файлы с папкой
  2. df1, сохранить как таблицу db или файл
...