(Py)SparkでJDBCソースを使用してデータの書き込みと読み込みを行うには？

質問

この質問の目的は、文書化することです。

• PySparkでJDBC接続を使用してデータを読み書きするために必要な手順

• JDBCソースで起こりうる問題とその解決方法

小さな変更で、これらのメソッドはScalaやRを含む他のサポートされる言語でも動作するはずです。

どのように解決するのですか？

データの書き込み

1. アプリケーションの送信時やシェルの起動時に、適用可能なJDBCドライバを含めるようにします。例えば --packages :

    bin/pyspark --packages group:name:version  
   
   

または組み合わせる driver-class-path と jars

    bin/pyspark --driver-class-path $PATH_TO_DRIVER_JAR --jars $PATH_TO_DRIVER_JAR

これらのプロパティは PYSPARK_SUBMIT_ARGS 環境変数を用いて設定することもできますし、JVMインスタンスが起動する前に conf/spark-defaults.conf を設定する spark.jars.packages または spark.jars / spark.driver.extraClassPath .

1. 希望のモードを選択します。Spark JDBC writerは以下のモードをサポートしています。

   • append : この :class の内容を追加します。 DataFrame の内容を既存のデータに追加します。
   • overwrite : 既存のデータを上書きします。
   • ignore : データが既に存在する場合、この操作を黙って無視します。
   • error (デフォルトケース): データが既に存在する場合は例外を投げる。

   アップサートやその他の細かい変更 はサポートされていません。

    mode = ...
   
   

2. JDBCのURIを用意する。

    # You can encode credentials in URI or pass
    # separately using properties argument
    # of jdbc method or options
   
    url = "jdbc:postgresql://localhost/foobar"
   
   

3. (オプション) JDBC引数の辞書を作成します。

    properties = {
        "user": "foo",
        "password": "bar"
    }
   
   

   properties / options を設定するために使用することもできます。 サポートされる JDBC 接続のプロパティ .

4. 使用方法 DataFrame.write.jdbc

    df.write.jdbc(url=url, table="baz", mode=mode, properties=properties)
   
   

を使って保存します( pyspark.sql.DataFrameWriter を参照)。

既知の問題 :

• を使用してドライバをインクルードした場合、適切なドライバが見つかりませんでした。 --packages ( java.sql.SQLException: No suitable driver found for jdbc: ... )

  ドライバのバージョンの不一致がないと仮定して、これを解決するために driver クラスを properties . 例えば

    properties = {
        ...
        "driver": "org.postgresql.Driver"
    }
  
  

• を使って df.write.format("jdbc").options(...).save() になる可能性があります。

  java.lang.RuntimeException: org.apache.spark.sql.execution.datasources.jdbc.DefaultSource は select としてテーブルを作成することを許可しません。

  解決策不明です。

• Pyspark 1.3では、Javaのメソッドを直接呼び出してみることができます。

    df._jdf.insertIntoJDBC(url, "baz", True)
  
  

データの読み込み

1. のステップ1～4に従ってください。 データを書き込む

2. 使用方法 sqlContext.read.jdbc :

    sqlContext.read.jdbc(url=url, table="baz", properties=properties)
   
   

または sqlContext.read.format("jdbc") :

    (sqlContext.read.format("jdbc")
        .options(url=url, dbtable="baz", **properties)
        .load())

既知の問題とゴチャゴチャ :

• 適切なドライバが見つかりません - 参照。データの書き込み

• Spark SQLはJDBCソースによる述語のプッシュダウンをサポートしていますが、すべての述語をプッシュダウンできるわけではありません。また、制限や集約を委譲することはできません。可能な回避策は dbtable / table 引数に有効なサブクエリを指定します。例えば、以下を参照してください。

  • spark述語プッシュダウンはJDBCで動作しますか？
  • pyspark.sql.DataFrame.take(4)の実行に1時間以上かかる。
  • dbtableでテーブルを定義するためにSQLクエリを使用するには？

• デフォルトでは、JDBC データソースは、単一の実行スレッドを使用してデータを順次ロードします。分散データロードを確実にするために、次のことができます。

  • パーティショニングを提供する column (必ず IntegerType ), lowerBound , upperBound , numPartitions .
  • 相互に排他的な述語のリストを提供する predicates のリストを提供し、希望する各パーティションに1つずつ提供する。

  参照してください。

  • RDBMS から JDBC 経由で読み込む際の spark でのパーティショニング ,
  • JDBC ソースからデータを移行する際にパーティショニングを最適化する方法は？ ,
  • DataFrameとJDBC接続を使用して遅いSparkジョブのパフォーマンスを向上させる方法は？
  • JDBCを使用してPostgresをインポートする場合、Spark RDDをどのようにパーティションするか？

• 分散モード（パーティショニングカラムや述語を使用）では、各エクゼキュータはそれ自身のトランザクションで動作します。ソースデータベースが同時に変更された場合、最終的なビューの一貫性が保たれる保証はありません。

適切なドライバを見つける場所。

• Maven リポジトリ (に必要な座標を取得するため --packages を選択し、Gradle タブからフォームにデータをコピーします。 compile-group:name:version をそれぞれのフィールドに置き換えてください）または Mavenセントラルリポジトリ :

  • PostgreSQL
  • MySQL

その他のオプション

データベースによっては、特別なソースが存在し、場合によっては優先されるかもしれません。

• グリーンプラム Pivotal Greenplum-Spark コネクタ
• Apache Phoenix - (英語) Apache Spark プラグイン
• マイクロソフト SQL サーバー Azure SQL データベースおよび SQL Server 用の Spark コネクタです。
• Amazon Redshift - （英語 Databricks Redshift コネクタ (現在のバージョンは、独自のDatabricks Runtimeでのみ利用可能です。 廃止されたオープンソース版、GitHubで利用可能 ).