Spark中Mysql数据处理

Question

我有一个要求，我需要每 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()

Answer 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