Spark a query from PostgreSQL(JDBC)中的分区

Question

我正在集群模式下运行spark（使用pyspark）并通过JDBC从RDBMS读取数据。我通过查询（而不是直接通过表）读取信息

我使用选项来分区，比如numPartitions、upperBound等。

sql = (select ... )

而且

df=spark
.read
.jdbc(url=jdbcUrl, table=sql, 
properties=connectionProperties, column="brand_id", lowerBound=1, 
upperBound=12000,numPartitions=10000 )

不幸的是，在生成查询的末尾，在WHERE子句上触发转换分区选项，因此PostGreSQL读取full表full而不使用索引！

我有一个这样的问题

SELECT "brand_id","brand_name","eq_ref_raw","oe","btimestamp" FROM 
  ( select  ... ) 
tab WHERE brand_id >= 5 AND brand_id < 6  

Answer 1

你试图做的事情在目前的Spark版本中似乎是不可能的。执行的查询构造为：

   val sqlText = s"SELECT $columnList FROM ${options.table} $myWhereClause"
    stmt = conn.prepareStatement(sqlText,
        ResultSet.TYPE_FORWARD_ONLY, ResultSet.CONCUR_READ_ONLY)

（请参阅org.apache.spark.sql.execution.datasources.jdbc.jdbcrdd#compute)

options.table对应于table属性中的（SELECT...）语句。

您能解释一下为什么需要从子查询加载列吗？如果在此子查询中进行联接或其他SQL操作，则始终可以“绕过”这些操作，并使用Spark SQL来完成这些操作（联接、SQL操作等等）。

编辑：

正如您所解释的，使用子查询的原因是JSONB提取。显然，它作为SQL本机操作会执行得更好，但是如果您想使用Spark来并行化您的处理，IMO，您需要在Spark级别声明JSON处理，如下所示：

CREATE TABLE jsonb_test (
  content jsonb
);

INSERT INTO jsonb_test (content) VALUES 
('{"first_name": "X", "last_name": "Y"}');

和代码：

val opts = Map("url" -> "jdbc:postgresql://127.0.0.1:5432/spark_jsonb_test",
  "dbtable" -> "jsonb_test", "user" -> "root", "password" -> "root",
  "driver" -> "org.postgresql.Driver")
val schema = StructType(Seq(
  StructField("first_name", StringType, true), StructField("last_name", StringType, true)
))
import sparkSession.implicits._
val personDataFrame = sparkSession.read
  .format("jdbc")
  .options(opts)
  .load()
  .withColumn("person", functions.from_json($"content", schema))

val extractedJsonNames = personDataFrame.collect.map(row => row.get(1).toString)

extractedJsonNames should have size 1
extractedJsonNames(0) shouldEqual "[X,Y]"

Spark支持贯穿PostgresDialect的JSONB字段，在其将DB类型转换为Catalyst类型的方法中，将JSONB视为StringType:

  private def toCatalystType(
  typeName: String,
  precision: Int,
  scale: Int): Option[DataType] = typeName match {
      case "bool" => Some(BooleanType)
      case "bit" => Some(BinaryType)
      case "int2" => Some(ShortType)
      case "int4" => Some(IntegerType)
      case "int8" | "oid" => Some(LongType)
      case "float4" => Some(FloatType)
      case "money" | "float8" => Some(DoubleType)
      case "text" | "varchar" | "char" | "cidr" | "inet" | "json" | "jsonb" | "uuid" =>
        Some(StringType)