Spark中的Mysql数据处理答案

【问题标题】：Mysql data processing in SparkSpark中的Mysql数据处理
【发布时间】：2016-10-08 21:19:29
【问题描述】：

我有一个要求，我需要每 5 分钟从多个源系统（Mysql 实例）获取数据，并使用其他一些数据（假设存在于 S3 中）加入和丰富它们。

我想在 Spark 中进行此处理以将我的执行分布到多个执行器。

主要问题是每次我在 Mysql 中进行查找时，我只想获取最新的记录（比如说 lastModifiedOn > 时间戳）。这种对 MySql 行的选择性获取如何有效地发生？这是我尝试过的：

val filmDf = sqlContext.read.format("jdbc")
  .option("url", "jdbc:mysql://localhost/sakila")
  .option("driver", "com.mysql.jdbc.Driver").option("dbtable", "film").option("user", "root").option("password", "")
  .load()

【问题讨论】：

你能用你尝试过的方法更新你的问题吗？
@eliasah Yes 会更新帖子。

标签： mysql apache-spark spark-dataframe

【解决方案1】：

您应该将 spark sql 与 jdbc 数据源一起使用。我给你举个例子。

val res = spark.read.jdbc(
      url = "jdbc:mysql://localhost/test?user=minty&password=greatsqldb",
      table = "TEST.table",
      columnName = "lastModifiedOn",
      lowerBound = lowerTimestamp,
      upperBound = upperTimestamp,
      numPartitions = 20,
      connectionProperties = new Properties()
    )

Apache Spark 测试套件中还有更多示例：https://github.com/apache/spark/blob/master/sql/core/src/test/scala/org/apache/spark/sql/jdbc/JDBCSuite.scala

【讨论】：

谢谢，你能告诉我这里的 numPartitions 是什么意思吗？还有connectionProperties？
结果将是一个带有 numPartitions 分区的 DataFrame。 Spark 将并行执行 numPartitions 查询以检索结果。示例：lowerBound=1，upperBound=10，numPartitions=2，Spark 将执行两个查询，第一个在 1 到 5 之间，第二个在 6 到 10 之间。
connectionProperties 是一个映射，用于将一些属性传递给数据库。取决于您的数据库，您可以使用它或不使用它。