你的 list_of_databases 没有并行化。要进行并行处理，应该使用 foreach 或是Spark赋予的东西。
打开emr中的concurrent选项并为每个表发送emr步骤，或者您可以使用spark的fair调度器，它可以在内部并行地执行作业，只需对代码进行少量修改。

我可以提出两个解决方案：
1简单的方法
一次向emr提交多个作业（每个db一个作业）。如果监控是个问题，只需将失败日志写入s3或hdfs即可。
2需要更改代码位
您可以尝试使用线程来并行化从每个db提取的数据。我可以展示一个如何做的示例，但是您可能需要做更多的更改来适应您的用例。
示例实现：

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') 财产。这将为每个数据库创建一个线程。如果要控制并行运行的线程数，请相应地修改for循环。
此外，如果要在程序中创建新作业，请将sparksession与参数一起传递。