如何跨 EMR 集群中的节点运行 python 代码

Question

我有一个 Amazon EMR 集群 - 30 个节点 我的 Python 代码如下所示 -

spark = SparkSession \
        .builder \
        .appName("App") \
        .config(conf=sparkConf) \
        .getOrCreate()

def fetchCatData(cat, tableName):
    df_gl = spark.sql("select * from {} where category = {}".format(tableName, cat))
    df_pandas = df_gl.select("*").toPandas()
    df_pandas.to_csv("/tmp/split/{}_{}.csv".format(tableName, cat))

catList = [14, 15, 63, 65, 74, 21, 23, 60, 79, 86, 107, 147, 196, 199, 200, 201, 229, 263, 265, 267, 328, 421, 468, 469,504]
tableList = ["Table1","Table2"
             ,"Table3",
             "Table4", "Table5", "Table6",
             "Table7"
             ]

def main(args):
    log4jLogger = spark._jvm.org.apache.log4j
    LOGGER = log4jLogger.LogManager.getLogger(__name__)

    for table in tableList:
        LOGGER.info("Starting Split for {}".format(table))
        dataLocation = "s3://test/APP/{}".format( table)
        df = spark.read.parquet(dataLocation)
        df = df.repartition("CATEGORY").cache()
        df.createOrReplaceTempView(table)
        for cat in catList:
            fetchGLData(cat, table)

我想解决以下问题-

1. 基本上我想读取我的镶木地板数据，将其按类别划分并将其作为 pandas 数据框存储在 csv 中。
2. 目前我正在按顺序运行，我想在 EMR 中的一个节点上运行每个类别并行运行它
3. 我尝试使用多处理，但对结果不满意。

在最短的时间内解决这个问题的最佳方法是什么？

Answer 1

不确定为什么要转换为 pandas 数据帧，但使用从 spark sql 创建的 spark 数据帧，您可以直接写入 csv。

但是，如果您希望将 csv 作为一个文件，则需要重新分区为 1，这不会使用所有节点。如果您不关心它生成多少文件，那么您可以重新分区数据帧以包含更多分区。然后每个分区将被节点处理并输出，直到所有分区都完成。

单个文件不使用所有节点（注意 .csv 将是包含实际 csv 的文件夹）

df_gl = spark.sql("select * from {} where category = {}".format(tableName, cat)) df_gl.repartition(1).write.mode("overwrite").csv("/tmp/split/{}_{}.csv".format(tableName, cat))

使用多个节点进行并行处理并输出为多个拆分文件（注意 .csv 将是包含实际 csv 的文件夹）

df_gl = spark.sql("select * from {} where category = {}".format(tableName, cat)).repartition(10) df_gl.write.mode("overwrite").csv("/tmp/split/{}_{}.csv".format(tableName, cat))