E-MapReduce:Lire et écrire dans HBase

Étape 1 : Obtenir les fichiers JAR du connecteur Spark HBase

Suivez les étapes ci-dessous pour obtenir les packages de dépendances requis. Pour connaître la compatibilité des versions entre Spark, Scala, Hadoop et HBase, consultez la documentation officielle du connecteur Spark HBase.

1. Compilez et empaquetez le connecteur.

   Compilez le connecteur Spark HBase en utilisant les versions de Spark, Scala, Hadoop et HBase présentes dans votre environnement. Cette opération génère les deux fichiers JAR principaux suivants :

   • hbase-spark-1.1.0-SNAPSHOT.jar

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

     Par exemple, exécutez la commande Maven suivante pour compiler et empaqueter le connecteur avec des versions spécifiques.

     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

     Si votre environnement utilise les mêmes versions (Spark 3.4.2, Scala 2.12.10, Hadoop 3.2.0 et HBase 2.4.9), vous pouvez utiliser les fichiers JAR précompilés suivants :

     • hbase-spark-1.1.0-SNAPSHOT.jar

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

2. Obtenez les dépendances HBase requises. Extrayez les packages suivants des dossiers lib/shaded-clients et lib/client-facing-thirdparty du répertoire d'installation de HBase. Dans cet exemple, la version de HBase est la 2.4.9.

   • hbase-shaded-client-2.4.9.jar

   • hbase-shaded-mapreduce-2.4.9.jar

   • slf4j-log4j12-1.7.30.jar

3. Téléchargez les cinq fichiers JAR vers OSS. Consultez la rubrique Téléchargement simple pour obtenir des instructions.

Étape 2 : Créer une connexion réseau

EMR Serverless Spark nécessite une connexion réseau pour accéder au service HBase du cluster HBase. Pour plus d'informations, consultez la rubrique Connecter EMR Serverless Spark à d'autres VPC.

Étape 3 : Créer une table HBase

1. Connectez-vous au cluster via SSH. Pour plus de détails, consultez la rubrique Se connecter à un cluster.

2. Exécutez la commande suivante pour vous connecter à HBase.

   hbase shell

3. Exécutez la commande suivante pour créer une table de test.

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

4. Exécutez les commandes suivantes pour écrire des données de test dans la 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'

Étape 4 : Lire une table HBase

1. Créez une session de notebook. Pour plus de détails, consultez la rubrique Gérer les sessions de notebook.

   Lors de la création de la session, sélectionnez la version du moteur correspondant à votre connecteur Spark HBase dans la liste déroulante Engine Version. Dans la section Normal Network Connection, sélectionnez la connexion réseau créée à l'étape 2. Ensuite, dans la section Spark Configuration, ajoutez les paramètres suivants pour charger le connecteur Spark HBase.

   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

   Le tableau suivant décrit ces paramètres.

   Paramètre	Description	Exemple
   spark.jars	Chemins OSS vers les fichiers JAR de dépendance externes.	Par exemple, l'un des cinq fichiers téléchargés vers OSS est oss://<yourBucketname>/spark/hbase/hbase-shaded-client-2.4.9.jar.
   spark.hadoop.hbase.zookeeper.quorum	Adresse IP interne du quorum ZooKeeper.	Si vous utilisez un autre cluster HBase, spécifiez la configuration appropriée. Si vous utilisez un cluster EMR HBase, vous trouverez l'Private IP Address du nœud master dans l'onglet Nodes du cluster.
   spark.hadoop.hbase.zookeeper.property.clientPort	Port du service ZooKeeper.	Si vous utilisez un autre cluster HBase, spécifiez le port approprié. Si vous utilisez un cluster Alibaba Cloud EMR HBase, le port est 2181.

2. Sur la page Development, créez une tâche de type Interactive Development > Notebook. Ensuite, dans le coin supérieur droit, sélectionnez la session de notebook que vous avez créée.

   Pour plus d'informations, consultez la rubrique Gérer les sessions de notebook.

3. Copiez le code suivant dans une nouvelle cellule du notebook, modifiez les paramètres si nécessaire, puis cliquez sur 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()
   

   Si les données attendues sont renvoyées, la configuration est correcte.

Étape 5 : Écrire dans une table HBase

Copiez le code suivant dans une nouvelle cellule du notebook, modifiez les paramètres si nécessaire, puis cliquez sur 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()

Après l'opération d'écriture, interrogez la table pour vérifier que les données ont bien été écrites.

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() La sortie inclut la nouvelle ligne r3 (ville=New York, âge=26, nom=sam), ce qui confirme que les données ont été écrites avec succès dans la table HBase.