Connector検証TweetPocketこんにちは。前回のブログでHive3の新しくなったACIDトランザクションを試しましたが、今回はそのトランザクションを有効にしたテーブルと、Spark2の連携について検証してみました。実際の運用では多くの場合、HiveとSpark両方を使うことが想定されるためです。環境とデータ前回のブログと全く同じ環境(GMOアプリクラウド&HDP3.0.1)で引き続き検証しました。使用するデータも同じKaggleのアイオワ州のお酒販売のデータです。参考までに、このデータの元のCSVのサイズは約3.2GBで1259万件ほどのレコードがあり、それをHive3のトランザクション管理テーブルに格納(ORCフォーマットに変換)したものを検証に使用しています。肝心のYARNとHiveのパラメータ及びYARNキューの設定は前回のブログと同じ設定です。Sparkについてのパラメータ設定については次の第2章で後述します。Hive3とSpark2の連携はどうやるのか?Spark2系になる前までは、SparkContextを使いその上でSQLContextかHiveContextを利用して、Hiveのテーブルに読み書きするのがメインでした。それがSpark2系になると、新たにSparkSessionができてこちらの利用が推奨されるようになりました。Hive3のトランザクションが有効になっているテーブルに対しては、そんなに単純な話ではなさそうな感じもします。Connectorというのが開発されて利用できるようになっていました。Connectorを利用した連携と、SparkSessionだけの連携の二つを行ってみました。検証概要お酒販売データのHiveテーブルをSparkのDataframeとして読み込み、日付を1年後の日付となるように加工し別のテーブルにSparkから書き込むということを試しました。Sessionだけで行うか、それぞれ試しました。ConnectorのJARファイルとPythonファイル(PySpark用)が最初から用意されています。検証では、データサイエンティストが実際に使うことを想定して、SparkをPythonで扱うPySparkを用いてSparkをクラスタモードで動かしました(spark-submit)。Connectorを使うには、追加でSpark2のパラメータ設定も必要になります。このあたりはこちらの記事にまとめられているので、一度読んでみてください。実際に追加したパラメータ設定内容は以下です。ほとんどは他のパラメータの値のコピーになります。jdbc:hive2://hdp3-m1.testhdp.com:2181,hdp3-m2.testhdp.com:2181,hdp3-m3.testhdp.com:2181/;serviceDiscoveryMode=zooKeeper;zooKeeperNamespace=hiveserver2-interactive上記パラメータを追加したら、Ambariが表示する再起動な必要なコンポーネントを再起動して、パラメータを反映します。今回の検証はsparkユーザーで行いました。結論から言うと、hiveユーザー以外ではHive3から通常のテーブルの格納先となった/warehouse/tablespace/managed/hive配下を見ることができなくて、パーミッションエラーで失敗します。自分ももちろんやらかしました。パーミッションエラーはこういう検証では必ず通る通過儀礼みたいなものですね。本来はきちんとユーザーとパーミッション管理を設計して行うべきものですが、今回はひとまずchmodして対応しました。sparkユーザーはhiveユーザーと同じhadoopグループに所属しているので、770のパーミッションを付与します。|+----------------------------------------------------+これで準備が整いました。PySparkシェルで試すこの手のものは、最初からクラスタモードで分散処理を行ったりするとハマるので(はい、ハマりました)、まずPySparkシェルで試してみます。PySparkシェルはインタラクティブにPySparkのプログラムを試すことができ、ローカルモードでやってみました。Connectorのファイルを指定します。Connectorが使えるようにします。以下のようにしてHiveとセッションを張ります。HiveWarehouseSession.session(spark).build()Hiveとのセッションが確立されたら、お酒販売データのテーブルがあるデータベースに接続して試しに件数を取得してみます。_c0|+--------+|12590429|+--------+ちゃんとHiveテーブルの件数がPySparkで取れました!あとはこのやり方をクラスタモードで分散処理できるようにすれば良いということになります。spark-submitでクラスタモードを試すクラスタモードで分散処理を行うには、spark-submitコマンドを使用します。PySparkを使うときもspark-submitコマンドを使用して、–py-filesオプションでPython(PySpark)ファイルを指定するだけです。検証したいことはHiveテーブルの読み込みだけでなく書き込みもなので、書き込み用のテーブルも作っておきます。'NO_AUTO_COMPACTION'='false');読み込み元のテーブルと全く同じ定義で、名前だけ変えたものです。Hiveに接続してテーブル作成しておきます。次に肝心のPySparkプログラムファイルを作ります。PySparkシェルで試した内容と全く同じように書いたら、最初全然うまくいかずハマりました・・。Connectorに”spark”という名前で引き渡しているのに、SparkSessionについて何も書いてないことに気が付きました・・。PySparkシェルでは最初からSparkSessionが”spark”という名前で引き渡せるようになっているみたいです。結局、以下のように書けばHiveとのセッションを確立できました。こちらの記事なども参考にしてみてください。HiveWarehouseSession.session(spark).build()繋がりさえすれば、あとはデータをDataframeに読み込んで分散処理して、Hiveテーブルに書き込むところを書き足します。検証で使ったPythonプログラムは以下のようになりました。"pyspark_iowa_liquor_sales").save()これを呼び出すspark-submitのシェルスクリプトは以下です。pyspark1.logこのpyspark1.sh(test_main1.py)をsparkユーザーで実行したところ、格納先のpyspark_iowa_liquor_salesテーブルに日付を加工した状態で無事に書き込めてました!クラスタモードで分散処理と言っても最初に試したのはexecutor数(num-executors)一つだけです。sparkはこのexecutor数を増やすことで同時に分散処理できる能力が上がります。各executorに割り当てるメモリサイズ(executor-memory)も大事で、処理性能に影響するだけでなく小さいとメモリ不足で処理が失敗したりします。Connectorの読み書きで色々とチューニング実験してみました。pyspark1.shの実行にかかった時間は2回平均で4分37秒ほどです。spark-submitのnum-executorsを4にpyspark1.shの内容からspark-submitのnum-executorsだけ1から4に変更して実行したところ、実行時間は2回平均で4分3秒ほどになりました。それほどでもないですが速くなりました。上記1の内容に加えて、Dataframeに読み込んだ後のパーティション数をexecutor数にあわせて4に、またcache()による一時キャッシュを利用するようにtest_main1.pyのコードを少し書き換えて試してみました。このためにPySparkを変更した箇所は以下です。"pyspark_iowa_liquor_sales").save()これで実行したところ、2回の実行時間の平均で6分27秒ほどでした。Connectorを使った書き込みで特に速くなることもなく、返って処理が増えて遅くなった印象です。上記2のrepartition(4)を(100)に変えてやってみたところ、実行時間が2回平均で5分38秒ほどでした。パーティション数を増やすとSparkが処理するタスクが増えますが、このケースでは上記2より速くなりました。上記1よりだいぶ遅くなっているので余計な処理に時間がかかってしまっている感じです。読み込み元のHiveテーブルについて、日付カラムを元に年と月のカラムを作り、年と月の2階層のパーティションを持つテーブルに変えてみました。その結果、71個のパーティションを持つHiveテーブルになったので、上記2の内容からまずcache()を外し、読み込み元のテーブルをパーティショニングされたテーブルに変更し、Dataframeのrepartition(71)を実施するように変えて試しました。会話していると時々混同してしまうときがありますが、Hiveテーブルのパーティショニングと、SparkのDataframeのパーティション数は全く違うものなので勘違いに注意です。以下のHiveQLでパーティショニングされたHiveテーブルを作成しました。iowa_liquor_sales;実行した結果、2回の実行時間の平均は3分37秒ほどになりました。読み込み元のテーブルをパーティショニングするのは大事というのが改めてわかる結果ですね。上記4の内容から、spark-submitのnum-executorを4から8に増やし、逆にexecutor-memoryを4GBから2GBに減らして実行してみました。結果、2回の実行時間の平均は2分47秒ほどでだいぶ速くなりました。WebUIを見るとどうも2Gのメモリに収まりきらず結局4G使っていて、今回のYARNのキュー設定ではdefaultキューのリソースが足りず、本来なら同時に8個立ち上がってほしいコンテナが5個しか立ち上がってないように見えました。今度は上記5の内容から、spark-submitのexecutor-memoryを2GBから1GBに減らしてやってみたところ、実行時間は2回平均で2分5秒ほどになりさらに速くなりました。Connectorを使っても問題なくHive3のトランザクションを有効にしたテーブルに読み書きできることはわかりました。Connectorを使わないでSpark2とHive3が連携できるか検証してみました。PySparkシェルで試すとにかく先ずはPySparkシェルでということで色々やってみたのですが、全然接続できなくて最もハマりました・・。Overflowのこのページを参考)。Connectorとは書き方が色々違います。途中データベースのリストを取得してみましたが、期待通りに取れています。データベースに接続できたところで続いてクエリを投げてみます。・・・・・sparkユーザーでPySparkシェルを呼び出してやったのですが、パーミッションエラーが出てしまいました。/warehouse/tablespace/managed/hiveのパーミッションを変えていてもSparkSessionをこのようなやり方で使うと怒られるようです。ここでは安直ですがsparkユーザーではなくhiveユーザーで行うことにしました。hiveユーザーで上記を実施したらエラーがなくなりました。次はSparkのクラスタモードを試します。spark-submitでクラスタモードを試すPySparkシェルで試した内容を元にPySparkプログラムとspark-submitシェルスクリプトを書いたのですが、HIVE_CONF_DIRの変数のセットがクラスタモードの分散処理で上手くいかずに手こずりました。以下のようにspark-submitにspark.yarn.appMasterEnv.HIVE_CONF_DIRを設定することで解決しました。pyspark14.logSparkSessionを使ったデータの書き込みは、新たにHiveの外部テーブルを作って試してみることにしました。'transactional'='false');この外部テーブルにSparkSessionからデータを書き込むようにしたPySparkプログラムが以下です。df.volume_sold_gallons)df.write.format("orc").mode("overwrite").saveAsTable("pyspark_iowa_liquor_sales_ex_no_trn")pyspark99.sh(test_main99.py)を実行したところ、エラーなく処理が完了しました!と思ったんですが、、明らかに処理時間が速く確認してみたところ、データがうまく保存できていないばかりか、勝手にテーブル定義が作り替わって(テーブルプロパティが書き替わって)しまうという事象が発生しました。エラーが出てくれないのかという感じです。作り替わってしまったテーブル定義は以下です。恐ろしいのは外部テーブルで作ったのに、Hiveのトランザクション管理テーブルになっているところです。|+----------------------------------------------------+明らかに今回のやり方は接続できたものの良くないと言えます。適切なやり方がもしかしたらある(もしくは今後対応する?)かもしれませんが、現時点ではSparkSessionのみでHive3のトランザクション管理テーブルに読み書きするのは難しそうです。Connectorを利用すべし、ということになりそうです。Connectorを使ってもSparkクラスタモードでの分散処理およびHiveテーブルへの読み書きは問題なくできました。Connectorを使ったSpark連携など、新しい部分を理解してシステムを見直さないといけないため、それまでのHadoopクラスタからは簡単には移行できなそうです。Hive3は魅力的な改善もありますけど。残る課題としては、Hive3のトランザクションを一切使わない(全て外部テーブルの)場合にSpark2とどのように連携するか?Cloudera社とHortonworks社が合併し、新しいHadoopクラスタパッケージのCDPではHiveとSparkの連携はどうなっていくか?このあたりはまたいつか試してみたいと思います。最後に次世代システム研究室では、データサイエンティスト/機械学習エンジニアを募集しています。ご応募をお願いします。皆さんのご応募をお待ちしています。