python之在(Pyspark

尝试将 postgreSQL DB 转换为 Dataframe。以下是我的代码:

from pyspark.sql import SparkSession 
 
spark = SparkSession.builder \ 
    .appName("Connect to DB") \ 
    .getOrCreate() 
 
jdbcUrl = "jdbc:postgresql://XXXXXX"  
connectionProperties = { 
  "user" : " ", 
  "password" : " ", 
  "driver" : "org.postgresql.Driver" 
} 
 
query = "(SELECT table_name FROM information_schema.tables) XXX" 
df = spark.read.jdbc(url=jdbcUrl, table=query, properties=connectionProperties) 
 
table_name_list = df.select("table_name").rdd.flatMap(lambda x: x).collect()  
    for table_name in table_name_list: 
          df2 = spark.read.jdbc(url=jdbcUrl, table=table_name, properties=connectionProperties) 

我得到的错误:

java.sql.SQLException: Unsupported type ARRAY on generating df2 for table name

如果我硬编码表名值，我不会得到同样的错误

df2 = spark.read.jdbc(jdbcUrl,"conditions",properties=connectionProperties)  

我检查了 table_name 类型，它是 String ，这是正确的方法吗？

请您参考如下方法:

我猜你不想要属于 postgres 内部工作的表名，例如 pg_type、pg_policies 等，它们的模式是 pg_catalog 导致错误的地方

py4j.protocol.Py4JJavaError: An error occurred while calling o34.jdbc. : java.sql.SQLException: Unsupported type ARRAY

当你试图把它们读成

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

并且有applicable_roles、view_table_usage等表，其架构类型为information_schema导致

py4j.protocol.Py4JJavaError: An error occurred while calling o34.jdbc. : org.postgresql.util.PSQLException: ERROR: relation "view_table_usage" does not exist

当你试图把它们读成

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

模式类型为public的表可以使用上述jdbc命令读入表。

I checked table_name type and it is String , is this the correct approach ?

因此您需要过滤掉那些表名并将您的逻辑应用为

from pyspark.sql import SparkSession 
 
spark = SparkSession.builder \ 
    .appName("Connect to DB") \ 
    .getOrCreate() 
 
jdbcUrl = "jdbc:postgresql://hostname:post/"  
connectionProperties = { 
  "user" : " ", 
  "password" : " ", 
  "driver" : "org.postgresql.Driver" 
} 
 
query = "information_schema.tables" 
df = spark.read.jdbc(url=jdbcUrl, table=query, properties=connectionProperties) 
 
table_name_list = df.filter((df["table_schema"] != 'pg_catalog') & (df["table_schema"] != 'information_schema')).select("table_name").rdd.flatMap(lambda x: x).collect()  
    for table_name in table_name_list: 
          df2 = spark.read.jdbc(url=jdbcUrl, table=table_name, properties=connectionProperties) 

这应该行得通