在 Pyspark 中将字典从一个笔记本传递到另一个笔记本时出现问题

Question

我是 pyspark 新手。我目前的项目需求是在Databricks中做ETL。我有一个 CSV 文件，其中有近 3 亿行，而这只是其中一个来源。将会有另外 2 个数据源。下面是我解决这个问题的方法：

Step1：创建抽象类和方法以从各种来源读取数据

Step2：读取step1中的数据并为每个源创建字典

第3步：将第2步中的字典传递到此步骤中并进行所有所需的转换

第四步：将数据加载到 parquet 文件中，然后加载到表中

我的问题是在步骤 3 中，我将使用从步骤 2 传递的字典。这可以吗，因为数据量太大，性能会很差。

请让我知道我应该采取什么方法，因为我陷入了步骤3。

提前谢谢您。

Answer 1

您可以直接使用 Spark DataFrames，而不是创建字典，因为它们分布在整个集群中，并且对它们的操作是并行的，这将显着提高性能。

例如，如果您按如下方式读取 CSV 文件：

df_csv = spark.read.csv("path/to/your/csv", header=True, inferSchema=True)

df_csv 是一个分布在集群中的 Spark DataFrame，可以高效处理大型数据集。

定义通用 ETL 方法

这可以对 Spark DataFrame 执行转换，此方法可以将 DataFrame 作为输入，应用必要的转换，并返回转换后的 DataFrame。

示例：

def transform_data(df):
    # Apply transformations like filtering, aggregating, joining, etc.
    df_transformed = df.filter(df["column_name"] > 100)  # Example transformation
    df_transformed = df_transformed.withColumn("new_column", df_transformed["column_name"] * 10)
    return df_transformed

然后可以将转换后的数据加载到 parquet 中，然后加载到表格中， 如果您有日期列，您可以使用它进行分区，这也有助于提高稍后阅读的性能。

df_transformed.write.partitionBy("date_column").parquet("path/to/output/directory", mode="overwrite")