在 PySpark SQL 中并行执行读写 API 调用

Question

我需要将 MySQL 中一组表的增量记录以 Parquet 格式加载到 Amazon S3。这些表在 AWS MySQL 托管实例中的多个数据库/模式中很常见。代码应该从每个模式（具有一组公共表）中并行复制数据。

我正在使用读取 API PySpark SQL 连接到 MySQL 实例并读取模式的每个表的数据，并使用写入 API 作为 Parquet 文件将结果数据帧写入 S3。我正在为数据库中的每个表循环运行它，如下面的代码所示：

def load_data_to_s3(databases_df):
    db_query_properties = config['mysql-query']
    auto_id_values = config['mysql-auto-id-values']
    for row in databases_df.collect():
        for table in db_query_properties.keys():
            last_recorded_id_value = auto_id_values[table]
            select_sql = "select * from {}.{} where id>{}".format(row.database_name, table, last_recorded_id_value)
            df = spark.read.format("jdbc") \
                    .option("driver", mysql_db_properties['driver']) \
                    .option("url", row.database_connection_url) \
                    .option("dbtable", select_sql) \
                    .option("user", username) \
                    .option("password", password) \
                    .load()
            s3_path = 's3a://{}/{}/{}'.format(s3_bucket, database_dir, table)
            df.write.parquet(s3_path, mode="append") 

我想知道如何将此代码扩展到在 EMR 集群中并行运行的多个数据库。请建议我一个合适的方法。如果需要更多详细信息，请告诉我。

Answer 1

我可以提出两种解决方案：

1.简单的方法

一次向您的 EMR 提交多个作业（每个数据库一个作业）。如果监控是问题，只需将失败日志写入 S3 或 HDFS。

2。需要更改一些代码

您可以尝试使用线程来并行化从每个数据库中提取的数据。我可以展示如何操作的示例，但您可能需要进行更多更改以适应您的用例。

示例实现：

import threading

def load_data_to_s3(databases_df):
    db_query_properties = config['mysql-query']
    auto_id_values = config['mysql-auto-id-values']
    for row in databases_df.collect():
        for table in db_query_properties.keys():
            last_recorded_id_value = auto_id_values[table]
            select_sql = "select * from {}.{} where id>{}".format(row.database_name, table, last_recorded_id_value)
            df = spark.read.format("jdbc") \
                    .option("driver", mysql_db_properties['driver']) \
                    .option("url", row.database_connection_url) \
                    .option("dbtable", select_sql) \
                    .option("user", username) \
                    .option("password", password) \
                    .load()
            s3_path = 's3a://{}/{}/{}'.format(s3_bucket, database_dir, table)
            df.write.parquet(s3_path, mode="append") 

threads = [threading.Thread(target=load_data_to_s3, args=(db) for db in databases_df]
    
for t in threads:
    t.start()

for t in threads:
    t.join() 

另外，请确保使用 set('spark.scheduler.mode', 'FAIR') 属性将调度程序更改为 FAIR。这将为您的每个数据库创建一个线程。如果要控制并行运行的线程数，请相应修改for循环。

此外，如果您想在程序中创建新作业，请将您的 SparkSession 与参数一起传递。

Answer 2

您的list_of_databases 未并行化。要进行并行处理，您应该并行化列表并使用 foreach 或 spark 提供的东西来执行并行工作。

---

在 EMR 中打开并发选项并为每个表发送 EMR 步骤，或者您可以使用 Spark 的公平调度程序，它可以在内部并行执行作业，只需对您的代码进行少量修改。