Spark：将大型 MySQL 表读入 DataFrame 失败

Question

我想提前告诉您，以下几个相关问题不能解决我的问题：

• Spark查询运行很慢
• 将mysql表转换为数据集非常慢...
• Spark 不会加载大型 MySql 表
• 从数据库读取时 Spark MySQL 错误

这个很接近，但堆栈跟踪不同，无论如何它都没有解决。所以请放心，我在几天（失败的）解决方案搜索后发布了这个问题。

---

我正在尝试编写一个作业，将数据（每天一次）从MySQL表移动到Hive存储为Parquet/ORC文件的表Amazon S3。一些表非常大：~ 300M 记录，大小为200 GB 以上（如 所报告phpMyAdmin）。

目前我们正在sqoop为此使用，但Spark由于以下原因，我们想转向：

• 为了利用它的功能DataFrame API（将来，我们将在移动数据的同时执行转换）
• 我们已经为组织中其他地方使用的工作编写了一个相当大的框架ScalaSpark

---

我已经能够在小 MySQL桌子上实现这一点而没有任何问题。但是，如果我尝试一次获取超过1.5-2M 条记录，那么Spark作业（从MySQLinto读取数据DataFrame）就会失败。我在下面展示了堆栈跟踪的相关部分，您可以在此处找到完整的堆栈跟踪。

...
javax.servlet.ServletException: java.util.NoSuchElementException: None.get
    at org.glassfish.jersey.servlet.WebComponent.serviceImpl(WebComponent.java:489)
    at org.glassfish.jersey.servlet.WebComponent.service(WebComponent.java:427)
...
Caused by: java.util.NoSuchElementException: None.get
    at scala.None$.get(Option.scala:347)
    at scala.None$.get(Option.scala:345)
...
org.apache.spark.status.api.v1.OneStageResource.taskSummary(OneStageResource.scala:62)
    at sun.reflect.GeneratedMethodAccessor188.invoke(Unknown Source)
    at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43)
    at java.lang.reflect.Method.invoke(Method.java:498)
...
[Stage 27:>                                                       (0 + 30) / 32]18/03/01 01:29:09 WARN TaskSetManager: Lost task 3.0 in stage 27.0 (TID 92, ip-xxx-xx-xx-xxx.ap-southeast-1.compute.internal, executor 6): java.sql.SQLException: Incorrect key file for table '/rdsdbdata/tmp/#sql_14ae_5.MYI'; try to repair it
    at com.mysql.jdbc.SQLError.createSQLException(SQLError.java:964)
    at com.mysql.jdbc.MysqlIO.checkErrorPacket(MysqlIO.java:3973)
...

**此堆栈跟踪是在移动包含186M记录的148 GB表失败时获得的

从（完整）堆栈跟踪中可以明显看出，Spark读取作业开始因错误的错误警告而生闷气，然后是（与的tmp 表变满有关）None.getSQLException: Incorrect key for file..MySQL

---

现在显然这不是MySQL问题，因为在这种情况下也sqoop应该失败。至于Spark而言，我并行读取操作通过设置numPartitions = 32（我们使用的40并行性sqoop）。

根据我对和 的有限了解，无论如何，148 GB对 Spark 来说都不应该是压倒性的。此外，因为，（）和所有驻留在相同的区域（），因此等待时间不应该是瓶颈。SparkBigDataMySQLSparkEMRS3AWS AP-SouthEast

---

我的问题是：

1. 是否Spark有合适的工具？
2. 能Spark的Jdbc 司机被指责为这个问题？
3. 如果以上问题的答案是
   • 是的：我怎样才能克服它？（替代驱动程序，或其他一些解决方法）？
   • 否：可能的原因是什么？

---

框架配置：

• Hadoop发行版：亚马逊 2.8.3
• Spark 2.2.1
• Hive 2.3.2
• Scala 2.11.11

EMR 配置：

• EMR 5.12.0
• 1 Master: r3.xlarge [8 vCore，30.5 GiB 内存，80 SSD GB 存储 EBS 存储：32 GiB]
• 1 Task: r3.xlarge [8 vCore，30.5 GiB 内存，80 SSD GB 存储 EBS 存储：无]
• 1 Core: r3.xlarge [8 vCore，30.5 GiB 内存，80 SSD GB 存储 EBS 存储：32 GiB]

**这些是开发集群的配置；生产集群配备更好

Answer 1

Spark JDBC API 似乎无需分叉即可将所有数据从 MySQL 表加载到内存中。因此，当您尝试加载大表时，您应该首先使用 Spark API 将数据克隆到 HDFS（应使用 JSON 来保留模式结构），如下所示：

spark.read.jdbc(jdbcUrl, tableName, prop)
       .write()
       .json("/fileName.json");

然后你就可以正常使用 HDFS 了。

spark.read().json("/fileName.json")
       .createOrReplaceTempView(tableName);