私はsbtとアセンブリを使用して、Sparkジョブ用にすべての依存関係を1つのjarにパッケージングしています。 私はいくつかのジョブで c3p0
を使って接続プールの情報を設定し、それをブロードキャストし、そして RDD で foreachPartition
を使って接続を取得し、データベースにデータをインサートしていました。 sbtのビルドスクリプトでは、次のように記述しています。
"mysql" % "mysql-connector-java" % "5.1.33"
これは、JDBCコネクタがジョブとともにパッケージングされていることを確認するものです。 すべてうまくいきました。
最近SparkSQLをいじり始めて、1.3.0
の新機能で単純にデータフレームを取得してjdbcソースに保存する方がずっと簡単だと気付きました。
次のような例外が発生します。
java.sql.SQLException:に対する適切なドライバが見つかりません。 jdbc:mysql://some.domain.com/myschema?user=user&password=password で。 java.sql.DriverManager.getConnection(DriverManager.java:596) at java.sql.DriverManager.getConnection(DriverManager.java:233)
ローカルで実行していたときは、この問題を回避するために
SPARK_CLASSPATH=/path/where/mysql-connector-is.jar
最終的に私が知りたいのは、ドライバと一緒にパッケージされているはずのジョブが、なぜドライバを見つけられないのか、ということです。 私の他のジョブにはこの問題はありませんでした。 私が言えることは、c3p0
とdataframeコードの両方が java.sql.DriverManager
(私が言えることは、あなたのためにすべてのインポートを処理する)を使用するので、それはうまくいくはずでしょうか? もし、アセンブリメソッドが動作するのを妨げる何かがあるのなら、これを動作させるために何をする必要があるのでしょうか?
この方は同じような問題を抱えていました。http://apache-spark-user-list.1001560.n3.nabble.com/How-to-use-DataFrame-with-MySQL-td22178.html
コネクタドライバを最新バージョンに更新しましたか?また、load()を呼び出す際に、ドライバクラスを指定しましたか?
Map<String, String> options = new HashMap<String, String>();
options.put("url", "jdbc:mysql://localhost:3306/video_rcmd?user=root&password=123456");
options.put("dbtable", "video");
options.put("driver", "com.mysql.cj.jdbc.Driver"); //here
DataFrame jdbcDF = sqlContext.load("jdbc", options);
spark/conf/spark-defaults.conf で spark.driver.extraClassPath と spark.executor.extraClassPath を MySql ドライバの .jar のパスに設定することも可能です。
この問題を解決する簡単なJavaのトリックが存在します。Class.forName()`のインスタンスを指定するのです。例えば
val customers: RDD[(Int, String)] = new JdbcRDD(sc, () => {
Class.forName("com.mysql.jdbc.Driver")
DriverManager.getConnection(jdbcUrl)
},
"SELECT id, name from customer WHERE ? < id and id <= ?" ,
0, range, partitions, r => (r.getInt(1), r.getString(2)))
docs]1をご確認ください。