将 MySQL 表转换为镶木地板时出现 Spark 异常答案

【问题标题】：Spark Exception when converting a MySQL table to parquet将 MySQL 表转换为镶木地板时出现 Spark 异常
【发布时间】：2017-03-10 11:37:26
【问题描述】：

我正在尝试使用 spark 1.6.2 将 MySQL 远程表转换为 parquet 文件。

该进程运行 10 分钟，内存已满，然后以以下消息开始：

WARN NettyRpcEndpointRef: Error sending message [message = Heartbeat(driver,[Lscala.Tuple2;@dac44da,BlockManagerId(driver, localhost, 46158))] in 1 attempts
org.apache.spark.rpc.RpcTimeoutException: Futures timed out after [10 seconds]. This timeout is controlled by spark.executor.heartbeatInterval

最后失败并出现此错误：

ERROR ActorSystemImpl: Uncaught fatal error from thread [sparkDriverActorSystem-scheduler-1] shutting down ActorSystem [sparkDriverActorSystem]
java.lang.OutOfMemoryError: GC overhead limit exceeded

我正在使用以下命令在 spark-shell 中运行它：

spark-shell --packages mysql:mysql-connector-java:5.1.26 org.slf4j:slf4j-simple:1.7.21 --driver-memory 12G

val dataframe_mysql = sqlContext.read.format("jdbc").option("url", "jdbc:mysql://.../table").option("driver", "com.mysql.jdbc.Driver").option("dbtable", "...").option("user", "...").option("password", "...").load()

dataframe_mysql.saveAsParquetFile("name.parquet")

我将最大执行程序内存限制为 12G。有没有办法强制将镶木地板文件写入“小”块以释放内存？

【问题讨论】：

您能否更具体地了解您的集群配置？
不幸的是它是独立的
你的数据有多大？任何分区列？
2600 万行，约 3GB 数据。没有分区
问题来了，没有分区！

标签： apache-spark apache-spark-sql parquet

【解决方案1】：

问题似乎在于您在使用 jdbc 连接器读取数据时没有定义分区。

默认情况下不分发从 JDBC 读取，因此要启用分发，您必须设置手动分区。您需要一个作为良好分区键的列，并且您必须预先了解分布情况。

这显然是您的数据的样子：

root 
|-- id: long (nullable = false) 
|-- order_year: string (nullable = false) 
|-- order_number: string (nullable = false) 
|-- row_number: integer (nullable = false) 
|-- product_code: string (nullable = false) 
|-- name: string (nullable = false) 
|-- quantity: integer (nullable = false) 
|-- price: double (nullable = false) 
|-- price_vat: double (nullable = false) 
|-- created_at: timestamp (nullable = true) 
|-- updated_at: timestamp (nullable = true)

order_year 对我来说似乎是个不错的人选。（根据您的 cmets，您似乎有大约 20 年）

import org.apache.spark.sql.SQLContext

val sqlContext: SQLContext = ???

val driver: String = ???
val connectionUrl: String = ???
val query: String = ???
val userName: String = ???
val password: String = ???

// Manual partitioning
val partitionColumn: String = "order_year"

val options: Map[String, String] = Map("driver" -> driver,
  "url" -> connectionUrl,
  "dbtable" -> query,
  "user" -> userName,
  "password" -> password,
  "partitionColumn" -> partitionColumn,
  "lowerBound" -> "0",
  "upperBound" -> "3000",
  "numPartitions" -> "300"
)

val df = sqlContext.read.format("jdbc").options(options).load()

PS： partitionColumn，lowerBound，upperBound，numPartitions：如果指定了其中任何一个选项，则必须全部指定这些选项。

现在您可以将您的DataFrame 保存到镶木地板。

【讨论】：