当Spark通过JDBC读取RDBMS时，是否有任何参数分区?

Question

当我运行 spark 应用程序进行表同步时，错误消息如下：

19/10/16 01:37:40 ERROR Executor: Exception in task 0.0 in stage 3.0 (TID 51)
com.mysql.cj.jdbc.exceptions.CommunicationsException: Communications link failure
The last packet sent successfully to the server was 0 milliseconds ago. The driver has not received any packets from the server.
at com.mysql.cj.jdbc.exceptions.SQLError.createCommunicationsException(SQLError.java:590)
at com.mysql.cj.jdbc.exceptions.SQLExceptionsMapping.translateException(SQLExceptionsMapping.java:57)
at com.mysql.cj.jdbc.ConnectionImpl.createNewIO(ConnectionImpl.java:1606)
at com.mysql.cj.jdbc.ConnectionImpl.<init>(ConnectionImpl.java:633)
at com.mysql.cj.jdbc.ConnectionImpl.getInstance(ConnectionImpl.java:347)
at com.mysql.cj.jdbc.NonRegisteringDriver.connect(NonRegisteringDriver.java:219)
at org.apache.spark.sql.execution.datasources.jdbc.DriverWrapper.connect(DriverWrapper.scala:45)
at org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$$anonfun$createConnectionFactory$1.apply(JdbcUtils.scala:63)
at org.apache.spark.sql.execution.datasources.jdbc.JdbcUtils$$anonfun$createConnectionFactory$1.apply(JdbcUtils.scala:54)
at org.apache.spark.sql.execution.datasources.jdbc.JDBCRDD.compute(JDBCRDD.scala:272)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
at org.apache.spark.rdd.MapPartitionsRDD.compute(MapPartitionsRDD.scala:52)
at org.apache.spark.rdd.RDD.computeOrReadCheckpoint(RDD.scala:324)
at org.apache.spark.rdd.RDD.iterator(RDD.scala:288)
at org.apache.spark.scheduler.ResultTask.runTask(ResultTask.scala:90)
at org.apache.spark.scheduler.Task.run(Task.scala:121)
at org.apache.spark.executor.Executor$TaskRunner$$anonfun$10.apply(Executor.scala:408)
at org.apache.spark.util.Utils$.tryWithSafeFinally(Utils.scala:1360)
at org.apache.spark.executor.Executor$TaskRunner.run(Executor.scala:414)
at java.util.concurrent.ThreadPoolExecutor.runWorker(ThreadPoolExecutor.java:1149)
at java.util.concurrent.ThreadPoolExecutor$Worker.run(ThreadPoolExecutor.java:624)
at java.lang.Thread.run(Thread.java:748)

我认为这是由表中的大量数据引起的。我之前用过 mongo 分区相关的参数，比如：spark.mongodb.input.partitioner、spark.mongodb.input.partitionerOptions.partitionSizeMB我想知道Spark在通过JDBC读取RDBMS时是否有类似的分区参数？

Answer 1

以下是参数及其描述，我们在使用 spark jdbc 读取 RDBMS 表时可以使用这些参数及其描述。

partitionColumn， lowerBound， upperBound- 如果指定了其中任何一个选项，则必须指定这些选项。此外，还必须指定 numPartitions。它们描述了在从多个工作线程并行读取时如何对表进行分区。partitionColumn 必须是相关表中的数字、日期或时间戳列。请注意，下限和上限仅用于确定分区步幅，而不用于筛选表中的行。因此，表中的所有行都将被分区并返回。此选项仅适用于阅读。

numPartitions- 可用于表读取和写入并行性的最大分区数。这也决定了并发 JDBC 连接的最大数量。如果要写入的分区数超过此限制，我们将在写入前调用 coalesce(numPartitions( 将其减少到此限制。

fetchsize- JDBC 提取大小，它确定每次往返要读取的行数。这可以帮助默认为低提取大小的 JDBC 驱动程序的性能(例如。甲骨文有 10 行(。此选项仅适用于阅读。

请注意，上述所有参数应一起使用。下面是一个例子：-

spark.read.format("jdbc").
option("driver", driver).
option("url",url ).
option("partitionColumn",column name).
option("lowerBound", 10).
option("upperBound", 10000).
option("numPartitions", 10).
option("fetchsize",1000).
option("dbtable", query).
option("user", user).
option("password",password).load()