如何跨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. 基本上我想读取我的镶木地板数据，按类别划分并将其存储为csv中的pandas数据帧。
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))