E-MapReduce:Read from and write to HBase

Step 1: Get the HBase Spark connector JARs

Obtain the required dependency packages by following these steps. For version compatibility between Spark, Scala, Hadoop, and HBase, refer to the official HBase Spark connector documentation.

1. Compile and package the connector.

   Use the versions of Spark, Scala, Hadoop, and HBase from your environment to compile the HBase Spark connector. This process generates the following two core JARs:

   • hbase-spark-1.1.0-SNAPSHOT.jar

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

     For example, use the following Maven command to compile and package the connector with specific versions.

     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

     If your environment uses the same versions (Spark 3.4.2, Scala 2.12.10, Hadoop 3.2.0, and HBase 2.4.9), you can use the following pre-compiled JARs:

     • hbase-spark-1.1.0-SNAPSHOT.jar

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

2. Obtain the required HBase dependencies. Extract the following dependency packages from the lib/shaded-clients and lib/client-facing-thirdparty folders in the HBase installation directory, where 2.4.9 is the HBase version number.

   • hbase-shaded-client-2.4.9.jar

   • hbase-shaded-mapreduce-2.4.9.jar

   • slf4j-log4j12-1.7.30.jar

3. Upload the five JARs to OSS. See Simple upload for instructions.

Step 2: Create a network connection

EMR Serverless Spark requires a network connection to access the HBase service in the HBase cluster. For more information about network connections, see Connect EMR Serverless Spark to other VPCs.

Step 3: Create an HBase table

1. Connect to the cluster using SSH. For details, see Log on to a cluster.

2. Run the following command to connect to HBase.

   hbase shell

3. Run the following command to create a test table.

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

4. Run the following commands to write test data to the table.

   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'

Step 4: Read an HBase table

1. Create a notebook session. For details, see Manage notebook sessions.

   When you create the session, select the engine version that corresponds to your HBase Spark connector from the Engine Version drop-down list. In the Normal Network Connection section, select the network connection that you created in Step 2. Then, in the Spark Configuration section, add the following parameters to load the HBase Spark connector.

   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 internal IP address
   spark.hadoop.hbase.zookeeper.property.clientPort  Zookeeper service port

   The following table describes these parameters.

   Parameter	Description	Example
   spark.jars	The OSS paths to the external dependency JARs.	For example, one of the five files uploaded to OSS is oss://<yourBucketname>/spark/hbase/hbase-shaded-client-2.4.9.jar.
   spark.hadoop.hbase.zookeeper.quorum	The internal IP address of the ZooKeeper quorum.	If you use a different HBase cluster, specify the appropriate configuration. If you use an EMR HBase cluster, you can find the Private IP Address of the master node on the Nodes tab of the cluster.
   spark.hadoop.hbase.zookeeper.property.clientPort	The ZooKeeper service port.	If you use a different HBase cluster, specify the appropriate port. If you use an Alibaba Cloud EMR HBase cluster, the port is 2181.

2. On the Development page, create a task of type Interactive Development > Notebook. Then, in the upper-right corner, select the notebook session that you created.

   For more information, see Manage notebook sessions.

3. Copy the following code into a new cell in the notebook, modify the parameters as needed, and then click Run.

   # Read data from the HBase table.
   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()
   # Register a temporary view.
   df.createOrReplaceTempView("hbase_table_view")
   # Query the data by using SQL.
   results = spark.sql("SELECT * FROM hbase_table_view")
   results.show()
   

   If the expected data is returned, your configuration is correct.

Step 5: Write to an HBase table

Copy the following code into a new cell in the notebook, modify the parameters as needed, and then click Run.

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()

After writing the data, query the table to verify that the operation was successful.

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()