Pyspark JDBC 中的谓词不进行分区读取答案

【问题标题】：Predicate in Pyspark JDBC does not do a partitioned readPyspark JDBC 中的谓词不进行分区读取
【发布时间】：2021-03-05 11:32:35
【问题描述】：

我正在尝试使用 JDBC 读取在 PySpark 中读取 Mysql 表。这里棘手的部分是表相当大，因此会导致我们的 Spark 执行器在对表进行非分区普通读取时崩溃。

因此，目标函数基本上是我们想要对表进行分区读取。我们一直在尝试的几件事 -

我们查看了“numPartitions-partitionColumn-lowerBound-upperBound”组合。这对我们不起作用，因为我们的原始表的索引键是一个字符串，这仅适用于整数类型。
文档中建议的另一种选择是谓词选项。这似乎对我们不起作用，因为分区数似乎仍然是 1，而不是我们发送的谓词数。

我们使用的代码sn-p如下-

input_df = self._Flow__spark.read \
            .format("jdbc") \
            .option("url", url) \
            .option("user", config.user) \
            .option("password", config.password) \
            .option("driver", "com.mysql.cj.jdbc.Driver") \
            .option("dbtable", "({}) as query ".format(get_route_surge_details_query(start_date, end_date))) \
            .option("predicates", ["recommendation_date = '2020-11-14'",
                                   "recommendation_date = '2020-11-15'",
                                   "recommendation_date = '2020-11-16'",
                                   "recommendation_date = '2020-11-17'",
                                   ]) \
            .load()

它似乎在进行全表扫描（非分区），同时完全忽略了传递的谓词。在这方面获得一些帮助会很棒。

【问题讨论】：

我认为谓词不能成为一种选择。您必须改用 jdbc 函数。 github.com/apache/spark/blob/v2.4.4/sql/core/src/main/scala/org/…

标签： apache-spark jdbc pyspark apache-spark-sql partition

【解决方案1】：

尝试以下方法：

spark_session\
  .read\
  .jdbc(url=url,
        table= "({}) as query ".format(get_route_surge_details_query(start_date, end_date)),
        predicates=["recommendation_date = '2020-11-14'",
                    "recommendation_date = '2020-11-15'",
                    "recommendation_date = '2020-11-16'",
                    "recommendation_date = '2020-11-17'"],
        properties={
          "user": config.user,
          "password": config.password,
          "driver": "com.mysql.cj.jdbc.Driver"
        }
)

验证分区

df.rdd.getNumPartitions() # Should be 4

我在https://spark.apache.org/docs/latest/api/python/pyspark.sql.html?highlight=jdbc#pyspark.sql.DataFrameReader.jdbc 挖掘文档后发现了这个

【讨论】：