Spark-on-HBase：DataFrameベースのHBaseコネクター

使用法

以下に、コネクタの基本的な使用方法を示します。 Avroや複合キーのサポートなどの詳細と高度なユースケースについては、リポジトリの例を参照してください。

1）スキーママッピングのカタログを定義します：

[code language="scala"]def catalog = s"""{
         |"table":{"namespace":"default", "name":"table1"},
         |"rowkey":"key",
         |"columns":{
           |"col0":{"cf":"rowkey", "col":"key", "type":"string"},
           |"col1":{"cf":"cf1", "col":"col1", "type":"boolean"},
           |"col2":{"cf":"cf2", "col":"col2", "type":"double"},
           |"col3":{"cf":"cf3", "col":"col3", "type":"float"},
           |"col4":{"cf":"cf4", "col":"col4", "type":"int"},
           |"col5":{"cf":"cf5", "col":"col5", "type":"bigint"},
           |"col6":{"cf":"cf6", "col":"col6", "type":"smallint"},
           |"col7":{"cf":"cf7", "col":"col7", "type":"string"},
           |"col8":{"cf":"cf8", "col":"col8", "type":"tinyint"}
         |}
       |}""".stripMargin
[/code]

2）データを準備し、HBaseテーブルにデータを入力します：
case class HBaseRecord（col0：String、col1：Boolean、col2：Double、col3：Float、col4：Int、col5：Long、col6：Short、col7：String、col8：Byte）

object HBaseRecord {def apply（i：Int、t：String）：HBaseRecord ={val s =s””” row $ {“％03d” .format（i）}””” HBaseRecord（s、i％2 ==0、i.toDouble、i.toFloat、i、i.toLong、i.toShort、s” String $ i：$ t”、i.toByte）}}

val data =（0〜255）.map {i => HBaseRecord（i、“ extra”）}

sc.parallelize（data）.toDF.write.options（
Map（HBaseTableCatalog.tableCatalog-> catalog、HBaseTableCatalog.newTable->“ 5”））
.format（“org.apache.spark。 sql.execution.datasources.hbase”）
.save（）

3）DataFrameをロードします：
def withCatalog（cat：String）：DataFrame ={
sqlContext
.read
.options（Map（HBaseTableCatalog.tableCatalog-> cat））
.format（ “ org.apache.spark.sql.execution.datasources.hbase”）
.load（）
}

val df =withCatalog（catalog）

4）統合言語クエリ：
val s =df.filter（（（$” col0” <=“ row050” &&$” col0”>“ row040”）||
$” col0” ===“ row005” ||
$” col0” ===“ row020” ||
$” col0” ===“ r20” ||
$” col0” <=“ row005”）&&
（$” col4″ ===1 ||
$” col4″ ===42））
.select（“ col0”、“ col1”、“ col4”）
s .show

5）SQLクエリ：
df.registerTempTable（ "table"）
sqlContext.sql（ "select count（col1）from table"）。show

Spark-Packageの構成

ユーザーは、Spark-on-HBaseコネクターを標準のSparkパッケージとして使用できます。パッケージをSparkアプリケーションに含めるには、次を使用します。

spark-shell、pyspark、またはspark-submit

> $ SPARK_HOME / bin / spark-shell –packages zhzhan：shc：0.0.11-1.6.1-s_2.10

ユーザーは、パッケージを依存関係としてSBTファイルに含めることもできます。形式はspark-package-name：version

spDependencies + =“ zhzhan / shc：0.0.11-1.6.1-s_2.10”

セキュアクラスターでの実行

Kerberos対応のクラスターで実行するには、HBaseトークンの取得と更新がSparkによって行われ、コネクターから独立しているため、ユーザーはHBase関連のjarファイルをクラスパスに含める必要があります。つまり、ユーザーは、kinitを使用するか、principal / keytabを指定することにより、通常の方法で環境を開始する必要があります。次の例は、yarn-clientモードとyarn-clusterモードの両方を使用して安全なクラスターで実行する方法を示しています。 SPARK_CLASSPATHは両方のモードに設定する必要があり、サンプルjarはSparkの単なるプレースホルダーであることに注意してください。

export SPARK_CLASSPATH =/ usr / hdp / current / hbase-client / lib / hbase-common.jar：/usr/hdp/current/hbase-client/lib/hbase-client.jar:/usr/hdp/current/hbase- client / lib / hbase-server.jar：/usr/hdp/current/hbase-client/lib/hbase-protocol.jar：/usr/hdp/current/hbase-client/lib/guava-12.0.1.jar

hrt_qaがヘッドレスアカウントであるとすると、ユーザーはkinitに対して次のコマンドを使用できます。

kinit -k -t /tmp/hrt_qa.headless.keytab hrt_qa

/usr/hdp/current/spark-client/bin/spark-submit –class org.apache.spark.sql.execution.datasources.hbase.examples.HBaseSource –マスターyarn-client –packages zhzhan：shc：0.0.11- 1.6.1-s_2.10 –num-executors 4 –driver-memory 512m –executor-memory 512m –executor-cores1/usr/hdp/current/spark-client/lib/spark-examples-1.6.1.2.4.2。 0-106-hadoop2.7.1.2.4.2.0-106.jar

/usr/hdp/current/spark-client/bin/spark-submit –class org.apache.spark.sql.execution.datasources.hbase.examples.HBaseSource –masteryarn-cluster –files / etc / hbase / conf / hbase -site.xml –packages zhzhan：shc：0.0.11-1.6.1-s_2.10 –num-executors 4 –driver-memory 512m –executor-memory 512m –executor-cores 1 / usr / hdp / current / spark- client / lib / spark-examples-1.6.1.2.4.2.0-106-hadoop2.7.1.2.4.2.0-106.jar

すべてをまとめる

HBaseがDataFrameレベルでSparkをサポートする方法の概要を説明しました。 DataFrame APIを使用すると、Sparkアプリケーションは、他のデータソースに格納されているデータと同じくらい簡単にHBaseテーブルに格納されているデータを処理できます。この新機能により、HBaseテーブルのデータは、Sparkアプリケーションやその他のインタラクティブツール（例：ユーザーは、Spark内のHBaseテーブルに対して複雑なSQLクエリを実行したり、Dataframeに対してテーブル結合を実行したり、SparkStreamingと統合してより複雑なシステムを実装したりできます。

次は何ですか？

現在、コネクタはHortonworksリポジトリでホストされており、Sparkパッケージとして公開されています。現在、ApacheHBaseトランクに移行中です。移行中に、HBaseトランクのいくつかの重大なバグを特定しました。これらのバグは、マージとともに修正される予定です。コミュニティの作業は、ScanおよびBulkGetの基盤となるコンピューティングアーキテクチャを最適化するためのHBASE-14795およびHBASE-14796を含む包括的なHBase JIRA HBASE-14789、使いやすさのためのJSONユーザーインターフェイスを提供するためのHBASE-14801、 DataFrame書き込みパス、Avroサポート用のHBASE-15334、short、int、long、float、doubleなどのJavaプリミティブタイプをサポートするためのHBASE-15333、複合行キーをサポートするためのHBASE-15335、およびHBASE-15572オプションのタイムスタンプセマンティクスを追加します。コネクタの操作をさらに簡単にするコネクタの将来のバージョンを作成することを楽しみにしています。

謝辞

Hamel Kothari、Sudarshan Kadambi、Bloombergの各チームがこの作業を指導し、この作業の検証を支援してくれたことに感謝します。また、フィードバックを提供し、これを改善してくれたHBaseコミュニティにも感謝します。最後に、この作業は以前のSpark HBase統合からの教訓を活用しており、開発者の道を開いてくれたことに感謝します。

参照：

SHC：https://github.com/hortonworks/shc-release

Spark-package：http://spark-packages.org/package/zhzhan/shc

Apache HBase：https://hbase.apache.org/

Apache Spark：http://spark.apache.org/