E-MapReduce:HBase の読み取りと書き込み

ステップ 1：HBase Spark コネクタの JAR の取得

次の手順に従って、必要な依存パッケージを取得します。Spark、Scala、Hadoop、HBase 間のバージョンの互換性については、公式の HBase Spark コネクタドキュメントをご参照ください。

1. コネクタのコンパイルとパッケージ化

   ご利用の環境の Spark、Scala、Hadoop、HBase のバージョンを使用して、HBase Spark コネクタをコンパイルします。このプロセスにより、次の 2 つのコア JAR が生成されます：

   • hbase-spark-1.1.0-SNAPSHOT.jar

   • hbase-spark-protocol-shaded-1.1.0-SNAPSHOT.jar

     たとえば、次の Maven コマンドを使用して、特定のバージョンでコネクタをコンパイルおよびパッケージ化します。

     mvn -Dspark.version=3.4.2 -Dscala.version=2.12.10 -Dhadoop-three.version=3.2.0 -Dscala.binary.version=2.12 -Dhbase.version=2.4.9 clean package -DskipTests

     ご利用の環境で同じバージョン (Spark 3.4.2、Scala 2.12.10、Hadoop 3.2.0、HBase 2.4.9) を使用している場合は、次のプリコンパイル済み JAR を使用できます：

     • hbase-spark-1.1.0-SNAPSHOT.jar

     • hbase-spark-protocol-shaded-1.1.0-SNAPSHOT.jar

2. 必要な HBase 依存関係の取得。HBase のインストールディレクトリにある lib/shaded-clients および lib/client-facing-thirdparty フォルダから、次の依存パッケージを抽出します。ここで、2.4.9 は HBase のバージョン番号です。

   • hbase-shaded-client-2.4.9.jar

   • hbase-shaded-mapreduce-2.4.9.jar

   • slf4j-log4j12-1.7.30.jar

3. 5 つの JAR を OSS にアップロードします。手順については、「シンプルアップロード」をご参照ください。

ステップ 2：ネットワーク接続の作成

EMR Serverless Spark が HBase クラスター内の HBase サービスにアクセスするには、ネットワーク接続が必要です。ネットワーク接続の詳細については、「EMR Serverless Spark を他の VPC に接続する」をご参照ください。

ステップ 3：HBase テーブルの作成

1. SSH を使用してクラスターに接続します。詳細については、「クラスターへのログオン」をご参照ください。

2. 次のコマンドを実行して HBase に接続します。

   hbase shell

3. 次のコマンドを実行してテストテーブルを作成します。

   create 'hbase_table', 'c1', 'c2'

4. 次のコマンドを実行して、テストデータをテーブルに書き込みます。

   put 'hbase_table', 'r1', 'c1:name', 'Alice'
   put 'hbase_table', 'r1', 'c1:age', '25'
   put 'hbase_table', 'r1', 'c2:city', 'New York'
   put 'hbase_table', 'r2', 'c1:name', 'Bob'
   put 'hbase_table', 'r2', 'c1:age', '30'
   put 'hbase_table', 'r2', 'c2:city', 'San Francisco'

ステップ 4：HBase テーブルの読み取り

1. ノートブックセッションを作成します。詳細については、「ノートブックセッションの管理」をご参照ください。

   セッションを作成する際、Engine Version ドロップダウンリストから、お使いの HBase Spark コネクタに対応するエンジンバージョンを選択します。Normal Network Connection セクションで、ステップ 2 で作成したネットワーク接続を選択します。次に、Spark Configuration セクションで、次のパラメーターを追加して HBase Spark コネクタをロードします。

   spark.jars                                        oss://<bucketname>/path/to/hbase-shaded-client-2.4.9.jar,oss://<bucketname>/path/to/hbase-shaded-mapreduce-2.4.9.jar,oss://<bucketname>/path/to/hbase-spark-1.1.0-SNAPSHOT.jar,oss://<bucketname>/path/to/hbase-spark-protocol-shaded-1.1.0-SNAPSHOT.jar,oss://<bucketname>/path/to/slf4j-log4j12-1.7.30.jar
   spark.hadoop.hbase.zookeeper.quorum               Zookeeper の内部 IP アドレス
   spark.hadoop.hbase.zookeeper.property.clientPort  Zookeeper サービスポート

   次の表に、これらのパラメーターを示します。

   パラメーター	説明	例
   spark.jars	外部依存関係 JAR の OSS パス。	たとえば、OSS にアップロードされた 5 つのファイルのうちの 1 つは oss://<yourBucketname>/spark/hbase/hbase-shaded-client-2.4.9.jar です。
   spark.hadoop.hbase.zookeeper.quorum	ZooKeeper クォーラムの内部 IP アドレス。	別の HBase クラスターを使用する場合は、適切な設定を指定します。 EMR HBase クラスターを使用する場合、クラスターの Nodes タブでマスターノードの Private IP Address を確認できます。
   spark.hadoop.hbase.zookeeper.property.clientPort	ZooKeeper サービスポート。	別の HBase クラスターを使用する場合は、適切なポートを指定します。 Alibaba Cloud EMR HBase クラスターを使用する場合、ポートは 2181 です。

2. Development ページで、[インタラクティブ開発] > [ノートブック] タイプのタスクを作成します。次に、右上隅で、作成したノートブックセッションを選択します。

   詳細については、「ノートブックセッションの管理」をご参照ください。

3. 次のコードをノートブックの新しいセルにコピーし、必要に応じてパラメーターを変更してから、実行 をクリックします。

   # HBase テーブルからデータを読み取ります。
   df = spark.read.format("org.apache.hadoop.hbase.spark") \
       .option("hbase.columns.mapping", "id STRING :key, name STRING c1:name, age STRING c1:age, city STRING c2:city") \
       .option("hbase.table", "hbase_table") \
       .option("hbase.spark.pushdown.columnfilter", False) \
       .load()
   # 一時ビューを登録します。
   df.createOrReplaceTempView("hbase_table_view")
   # SQL を使用してデータをクエリします。
   results = spark.sql("SELECT * FROM hbase_table_view")
   results.show()
   

   期待されるデータが返された場合、設定は正しいです。

ステップ 5：HBase テーブルへの書き込み

次のコードをノートブックの新しいセルにコピーし、必要に応じてパラメーターを変更してから、実行 をクリックします。

from pyspark.sql.types import StructType, StructField, StringType
data = [
    ("r3", "sam", "26", "New York")
]
schema = StructType([
    StructField("id", StringType(), True),
    StructField("name", StringType(), True),
    StructField("age", StringType(), True),
    StructField("city", StringType(), True)
])
testDS = spark.createDataFrame(data=data,schema=schema)
testDS.write.format("org.apache.hadoop.hbase.spark").option("hbase.columns.mapping", "id STRING :key, name STRING c1:name, age STRING c1:age, city STRING c2:city").option("hbase.table", "hbase_table").save()

データを書き込んだ後、テーブルをクエリして操作が成功したことを確認します。

df = spark.read.format("org.apache.hadoop.hbase.spark") \
    .option("hbase.columns.mapping", "id STRING :key, name STRING c1:name, age STRING c1:age, city STRING c2:city") \
    .option("hbase.table", "hbase_table") \
    .option("hbase.spark.pushdown.columnfilter", False) \
    .load()
df.createOrReplaceTempView("hbase_table_view")
results = spark.sql("SELECT * FROM hbase_table_view")
results.show()