如何使用 spark.read.jdbc 读取不同 Pyspark 数据帧中的多个文件

Posted 2023-04-17

【中文标题】如何使用 spark.read.jdbc 读取不同 Pyspark 数据帧中的多个文件

【英文标题】：How to Read multiple files in different Pyspark Dataframes using spark.read.jdbc

【发布时间】：2020-11-23 17:19:37

【问题描述】：

我有一个代码可以将多个文件 (>10) 读取到 Pyspark 中的不同数据帧中。但是，我想使用 for 循环和引用变量或类似的东西来优化这段代码。我的代码如下：

Features_PM = (spark.read
          .jdbc(url=jdbcUrl, table='Features_PM',
                properties=connectionProperties))

Features_CM = (spark.read
          .jdbc(url=jdbcUrl, table='Features_CM',
                properties=connectionProperties))

我尝试了类似的方法，但没有成功：

table_list = ['table1', 'table2','table3', 'table4']

for table in table_list:
     jdbcDF = spark.read \
         .format("jdbc") \
         .option("url", "jdbc:postgresql:dbserver") \
         .option("dbtable", "schema.".format(table)) \
         .option("user", "username") \
         .option("password", "password") \
         .load()

以上sn-p来源：https://community.cloudera.com/t5/Support-Questions/read-multiple-table-parallel-using-Spark/td-p/286498

任何帮助将不胜感激。谢谢

【问题讨论】：

在列表中获取该数据库的所有表名，现在创建一个通用函数并通过迭代列表来读取所有表名.. 这样您就可以拥有一个函数来读取所有表格...代码可重用性

有人可以帮我写代码吗。

【参考方案1】：

您可以使用以下代码来实现您的最终目标。您将获得一个数据框字典，其中键是表名，值是适当的数据框

def read_table(opts):
    return spark.read.format("jdbc").options(**opts).load()

table_list = ['table1', 'table2','table3', 'table4']



table_df_dict = table: read_table("url":"jdbc:postgresql:dbserver",
                                   "dbtable":"schema.".format(table),
                                   "user": "username",
                                   "password":"password")
                 for table in table_list