将大数据从 MySQL 加载到 Spark答案

【问题标题】：Loading large data from MySQL into Spark将大数据从 MySQL 加载到 Spark
【发布时间】：2021-06-01 19:18:47
【问题描述】：

寻找对 Spark 的理解...

我正在将大量数据从 MySQL 加载到 Spark 中，但它一直在死机 :-(

org.apache.spark.SparkException: Job aborted.
at org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelation$$anonfun$run$1.apply$mcV$sp(InsertIntoHadoopFsRelation.scala:156)
at org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelation$$anonfun$run$1.apply(InsertIntoHadoopFsRelation.scala:108)
at org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelation$$anonfun$run$1.apply(InsertIntoHadoopFsRelation.scala:108)
at org.apache.spark.sql.execution.SQLExecution$.withNewExecutionId(SQLExecution.scala:56)
at org.apache.spark.sql.execution.datasources.InsertIntoHadoopFsRelation.run(InsertIntoHadoopFsRelation.scala:108)

这是我的代码

val query =
  s"""
     (
      select 
      mod(act.AccountID, ${parts}) part,
      p.Value name, event.EventTime eventTime, act.AccountID accountID, act.UserGoal goalID,event.ActivityID activityID, id.CountryID countryID, arr.ConsumerID consumerID
      from DimIdentity as id
      join FactArrival as arr on  arr.IdentityID=id.IdentityID
      join FactActivityEvent as event on event.ArrivalID=arr.ArrivalID
      join DimAccount as act on  act.AccountID=event.AccountID
      join DimAccountRoleTypeMatch as role on role.AccountID=act.AccountID
      join DimDateTime as d on event.DateTimeID=d.DateTimeID
      join DimProperty as p on p.PropertyID=event.EventTypeID
      where
        id.Botness=0 and 
        d.DayOfYear>=${from} and d.DayOfYear<${to} and d.Year=${year} and
        (role.AccountRoleTypeID=1 or role.AccountRoleTypeID=2)
  ) a
  """.stripMargin

val events = sqlContext.read.format("jdbc").
  option("url", sqlURL).
  option("driver", "com.mysql.jdbc.Driver").
  option("useUnicode", "true").
  option("zeroDateTimeBehavior", "round").
  option("continueBatchOnError", "true").
  option("useSSL", "false").
  option("dbtable", query).
  option("user", sqlUser).
  option("password", sqlPassword).
  option("partitionColumn", "part").
  option("lowerBound", "0").
  option("upperBound", s"${parts - 1}").
  option("numPartitions", s"${parts}").
  load().as[Activity].toDF

请注意，我正在使用其他答案中推荐的 partitionColumn、lowerBound、upperBound、numPartitions

我尝试将分区从 4 设置为 512，但它总是死掉。从文件或Mongo中读取相同数量的数据没有问题。这是 MySQL 连接器的问题吗？有解决办法吗？

请注意，我找到了一个建议我避免使用 Spark 的答案，并将查询读入 HDFS 上的文件，然后加载文件

Multiple Partitions in Spark RDD

这真的是最好的方法吗？

【问题讨论】：

对不起，没有意识到我应该这样做。我会修复
如果解决方案适合您，您应该这样做。这既是对发布者的奖励，也是对其他用户的标志，这是一个有效的解决方案。

标签： mysql apache-spark dataframe

【解决方案1】：

这是我得到的答案......

对我来说，答案是避免 Spark 的 mysql-connection :-( 我发现很难避免分区导致的崩溃。mysql-connection 需要手动调整分区，并且不会产生任何增加速度。编写将数据读入大型文本文件的非 Spark 代码以及在文本文件上调用 Spark 更容易。Spark 非常适合大多数数据源，但不是 mysql……至少现在还没有

【讨论】：

谢谢你的回答，我也面临同样的问题，因为你试图从 mysql 数据库（源）和另一个表目标表从 hive 中获取大量数据，以便在两者之间进行比较，但我得到的只是是java.lang.OutOfMemoryError: GC overhead limit exceeded spark，您能否给我一个提示，告诉我如何编写将数据读入大文本文件的非 Spark 代码。

【解决方案2】：

您可以尝试在不使用动态分区进行读取的情况下增加提取大小。

sqlContext.read.options(options).jdbc(
url=sqlURL, table=query, columnName="part",
fetchSize=1000000,connectionProperties=new java.util.Properties())

【讨论】：

【解决方案3】：

您可以通过使用限制偏移量更改 sql 查询来读取数据。然后使用 shell 脚本使用 for 循环自动执行任务。这对我有用

【讨论】：