Как создать динамический c фрейм данных

Question

Я пытался создать фрейм данных, и причина, по которой я дал создать фрейм данных следующим образом, состоит в том, чтобы сделать его динамическим c, но выражение передается в виде строки, а команда exe c не удалось создать фрейм данных и присвоить переменную.

Вот мой код:

def fileReader(inputFileType,sourceFilePath,inputFileType):
 value ='true'
 header='header'


 a= "spark.read.option('"+header+"','"+value+"')."+inputFileType+"('"+sourceFilePath+"')"
 print(a)
 print(type(a))
 ds = exec(a)
 return 'True'

Answer 1

Вы можете попробовать файл конфигурации, как показано ниже:

import json
files = json.loads('{
  "fileA": {
    "header": "true",
    "inputFileType": "csv",
    "sourceFilePath": "path_to_fileA"
  },
  "fileB": {
    "header": "true",
    "inputFileType": "parquet",
    "sourceFilePath": "path_to_fileB"
  }
}')


df_dict = {}
for file in files:
    df_dict['file'] = spark.read.option('header',file["header"]).format(file["inputFileType"]).path(file["sourceFilePath"])

Затем вы получите словарь данных с разными форматами и путем к файлу.

Answer 2

Привет, я получил ответ ...

def fileReader(inputFileType,sourceFilePath,inputFileType):
  value ='true'
  header='header'
  a= "spark.read.option('"+header+"','"+value+"')."+inputFileType+" 
  ('"+sourceFilePath+"')"
  print(a)
  print(type(a))
  ds = eval(a)
  return 'True'

мотивом создания этой функции является динамическое создание фреймов данных с использованием различных форматов файлов, поддерживаемых pyspark. Теперь с помощью этой функции я могу создать фрейм данных любого типа форматов файлов, поддерживаемых pyspark, просто передав местоположение и формат файлов.

Я ценю всю помощь.

Answer 3

Я бы рекомендовал использовать pyspark вместо exec для строк, представляющих команды spark. У pyspark действительно хороший интерфейс для выполнения команд spark в python. Он также имеет удобную функцию csv в модуле pyspark.sql, которая загружает данные из CSV-файла в кадр данных искры . Ваш код будет выглядеть примерно так:

from pyspark.sql import SparkSession
spark =(SparkSession.builder
                    .appName(YOUR_APP_NAME)
                    .getOrCreate())
csv_df = spark.csv('path/to/csv')

Однако, если ваши данные хранятся в CSV, я подозреваю, что не хватает данных, чтобы действительно гарантировать использование spark, который имеет много накладных расходов и предназначен для обработки очень больших наборов данных. Я бы порекомендовал сначала попробовать pandas, , который также имеет удобный способ загрузки CSV-данных в фрейм данных , и переключаться только на спарк, если ваш код работает слишком медленно с pandas.