Comment puis-je me connecter à une base de données postgreSQL dans Apache Spark utilisation de scala?

Notre objectif consiste à exécuter en parallèle des requêtes SQL à partir de l'Étincelle travailleurs.

Construire le programme d'installation

Ajouter le connecteur JDBC et à la libraryDependencies dans build.sbt. J'ai seulement essayé cela avec MySQL, donc je vais utiliser que dans mes exemples, mais Postgres devrait être sensiblement le même.

libraryDependencies ++= Seq(
  jdbc,
  "mysql" % "mysql-connector-java" % "5.1.29",
  "org.apache.spark" %% "spark-core" % "1.0.1",
  //etc
)

Code

Lorsque vous créez le SparkContext vous dira quels pots pour copier les exécuteurs testamentaires. Inclure le connecteur jar. Une belle façon de le faire:

val classes = Seq(
  getClass,                   //To get the jar with our own code.
  classOf[mysql.jdbc.Driver]  //To get the connector.
)
val jars = classes.map(_.getProtectionDomain().getCodeSource().getLocation().getPath())
val conf = new SparkConf().setJars(jars)

Maintenant Étincelle est prêt à se connecter à la base de données. Chaque exécuteur testamentaire exécuter la partie de la requête, de sorte que les résultats sont prêtes pour le calcul distribué.

Il y a deux options pour cela. L'ancienne approche est d'utiliser org.apache.spark.rdd.JdbcRDD:

val rdd = new org.apache.spark.rdd.JdbcRDD(
  sc,
  () => {
    sql.DriverManager.getConnection("jdbc:mysql://mysql.example.com/?user=batman&password=alfred")
  },
  "SELECT * FROM BOOKS WHERE ? <= KEY AND KEY <= ?",
  0, 1000, 10,
  row => row.getString("BOOK_TITLE")
)

Consultez la documentation pour les paramètres. Brièvement:

• Vous avez le SparkContext.
• Ensuite une fonction qui crée la connexion. Ce sera appelée à chaque travailleur de se connecter à la base de données.
• Alors la requête SQL. Ce doit être similaire à l'exemple, et contiennent des espaces réservés pour le début et la fin de la clé.
• Ensuite, vous spécifiez la plage de touches (0 à 1000 dans mon exemple) et le nombre de partitions. La gamme sera réparti entre les partitions. Donc, un exécuteur testamentaire thread sera à la fin de l'exécution de SELECT * FROM FOO WHERE 0 <= KEY AND KEY <= 100 dans l'exemple.
• Et enfin, nous avons une fonction qui convertit les ResultSet en quelque chose. Dans l'exemple que nous avons de le convertir en un String, si vous vous retrouvez avec un RDD[String].

Depuis Apache Spark version 1.3.0 une autre méthode est disponible à travers le DataFrame de l'API. Au lieu de la JdbcRDD vous devez créer un org.apache.spark.sql.DataFrame:

val df = sqlContext.load("jdbc", Map(
  "url" -> "jdbc:mysql://mysql.example.com/?user=batman&password=alfred",
  "dbtable" -> "BOOKS"))

Voir https://spark.apache.org/docs/1.3.1/sql-programming-guide.html#jdbc-to-other-databases pour la liste complète des options (la clé de la gamme et le nombre de partitions peut être définie comme avec JdbcRDD).

Mises à jour

JdbcRDD ne prend pas en charge les mises à jour. Mais vous pouvez simplement les faire dans un foreachPartition.

rdd.foreachPartition { it =>
  val conn = sql.DriverManager.getConnection("jdbc:mysql://mysql.example.com/?user=batman&password=alfred")
  val del = conn.prepareStatement("DELETE FROM BOOKS WHERE BOOK_TITLE = ?")
  for (bookTitle <- it) {
    del.setString(1, bookTitle)
    del.executeUpdate
  }
}

(Ce qui crée une connexion par partition. Si c'est un problème, utilisez un pool de connexion!)

DataFrames charge les mises à jour par le biais de la createJDBCTable et insertIntoJDBC méthodes.