E-MapReduce:Lire et écrire dans Hologres

Exemple 1 : Session SQL

Lisez et écrivez dans Hologres en utilisant une session SQL.

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

   Lors de la création de la session, sélectionnez la connexion réseau créée à l'étape précédente dans la liste des connexions réseau. Dans la section Spark Configuration, ajoutez les paramètres suivants pour charger le hologres-connector-spark.

   # Add the hologres-connector JAR file (only required for versions before esr-4.8.0).
   spark.emr.serverless.user.defined.jars oss://<bucket>/hologres-connector-spark-3.x-<version>.jar
   
   # Configure the Hologres catalog.
   spark.sql.catalog.hologres_external_test_db com.alibaba.hologres.spark3.HoloTableCatalog
   spark.sql.catalog.hologres_external_test_db.username ***
   spark.sql.catalog.hologres_external_test_db.password ***
   spark.sql.catalog.hologres_external_test_db.jdbcurl jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/testdb

   Le tableau suivant décrit les paramètres.

   Paramètre	Exemple	Description
   spark.emr.serverless.user.defined.jars	oss://<bucket>/hologres-connector-spark-3.x-<version>.jar	Chemin d'accès au fichier JAR défini par l'utilisateur.
   spark.sql.catalog.hologres_external_test_db	com.alibaba.hologres.spark3.HoloTableCatalog	Configure une source de données Hologres en tant que catalogue externe dans Spark 3.x. Il s'agit d'une valeur fixe.
   spark.sql.catalog.hologres_external_test_db.username	LTAI******	AccessKey ID de votre compte Alibaba Cloud. Nous vous recommandons d'utiliser la gestion des secrets pour traiter les informations sensibles. Pour plus de détails, consultez Gérer les informations sensibles à l'aide de la gestion des secrets.
   spark.sql.catalog.hologres_external_test_db.password	mXYV******	AccessKey Secret de votre compte Alibaba Cloud. Nous vous recommandons d'utiliser la gestion des secrets pour traiter les informations sensibles. Pour plus de détails, consultez Gérer les informations sensibles à l'aide de la gestion des secrets.
   spark.sql.catalog.hologres_external_test_db.jdbcurl	jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/testdb	URL de connexion JDBC de l'instance Hologres.

   Vous pouvez personnaliser la partie hologres_external_test_db dans les noms des paramètres.

2. Sur la page Data Development, créez une tâche SparkSQL et sélectionnez la session SQL créée dans le coin supérieur droit.

   Pour plus d'informations, consultez Développement SparkSQL.

3. Copiez le code suivant dans le nouvel onglet SparkSQL et cliquez sur Run.

   -- Switch to the testdb database.
   USE hologres_external_test_db;
   -- Write data.
   INSERT INTO `public`.test VALUES ('1004','tom');
   -- Query data.
   SELECT * FROM `public`.test;

   

Exemple 2 : Tâche de streaming

Cet exemple PySpark lit les données depuis Kafka et les écrit dans Hologres en tant que tâche de streaming.

1. Dans cet exemple de code, remplacez les informations Kafka et la table Hologres par vos valeurs réelles.

   from pyspark.sql import SparkSession
   from pyspark.sql.functions import col
   
   # Configure your Kafka information.
   servers = "alikafka-serverless-cn-xxxxx-vpc.alikafka.aliyuncs.com:9092"  # Replace with your Kafka bootstrap servers.
   topic = "topic-name"  # Replace with your Kafka topic.
   
   # Create a SparkSession.
   spark = SparkSession.builder \
       .appName("test read kafka") \
       .getOrCreate()
   
   # Read the Kafka stream.
   df = spark \
       .readStream \
       .format("kafka") \
       .option("kafka.bootstrap.servers", servers) \
       .option("subscribe", topic) \
       .load()
   
   # Define a function to write to Hologres (called for each micro-batch).
   def write_to_hologres(batch_df, batch_id):
       print(f"Writing batch {batch_id} to Hologres...")
       batch_df.write \
           .format("hologres") \
           .mode("append") \
           .insertInto("hologres_external_test_db.public.test")  # Replace with your Hologres table.
   
   # Convert the key and value to strings and write the stream.
   query = df.selectExpr("CAST(key AS STRING)", "CAST(value AS STRING)") \
       .writeStream \
       .foreachBatch(write_to_hologres) \
       .outputMode("append") \
       .trigger(processingTime='30 seconds') \
       .start()
   
   # Wait for the streaming query to terminate (this blocks cell execution in a notebook).
   query.awaitTermination()

2. Téléversez le fichier.

   1. Sur la page Artifacts, cliquez sur Upload File.

   2. Dans la boîte de dialogue Upload File, cliquez sur la zone de téléchargement pour sélectionner le fichier Python de l'étape précédente, ou faites glisser le fichier dans la zone de téléchargement.

3. Créez et exécutez la tâche de streaming.

   1. Sur la page Development, cliquez sur l'icône (Créer).

   2. Dans la boîte de dialogue qui s'affiche, saisissez un Name, sélectionnez PySpark dans la liste Application (Streaming), puis cliquez sur OK.

   3. Dans le nouvel onglet de développement, configurez les paramètres suivants et laissez les autres avec leurs paramètres par défaut. Ensuite, cliquez sur Publish.

      Paramètre	Description
      Main Python Resources	Sélectionnez le fichier Python téléversé à l'étape précédente.
      Engine Version	Sélectionnez une version Spark compatible. Cet exemple utilise esr-4.6.0.
      Network Connection	Sélectionnez la connexion réseau créée à l'étape 2.
      Spark Configuration	# Add the hologres-connector JAR file (only required for versions before esr-4.8.0). spark.emr.serverless.user.defined.jars oss://<bucket>/test_script/hologres-connector-spark-3.x-1.5.6-jar-with-dependencies.jar # Configure the Hologres catalog. spark.sql.catalog.hologres_external_test_db com.alibaba.hologres.spark3.HoloTableCatalog spark.sql.catalog.hologres_external_test_db.username *** spark.sql.catalog.hologres_external_test_db.password *** spark.sql.catalog.hologres_external_test_db.jdbcurl jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/testdb Pour une description détaillée des paramètres, consultez Exemple 1 : Session SQL.

   4. Une fois la tâche publiée, cliquez sur Go to O&M. Sur la page qui s'ouvre, cliquez sur Start.

4. Vérifiez le résultat.

   1. Envoyez des messages à Kafka.

      

   2. Interrogez les données à l'aide de Spark SQL.

Exemple 3 : Session Notebook

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

   Lors de la création de la session, sélectionnez la connexion réseau créée à l'étape précédente dans la liste des connexions réseau. Dans la section Spark Configuration, ajoutez le paramètre suivant pour charger le hologres-connector-spark.

   # Add the hologres-connector JAR file (only required for versions before esr-4.8.0).
   spark.emr.serverless.user.defined.jars oss://<bucket>/hologres-connector-spark-3.x-<version>.jar

2. Sur la page Data Development, créez une tâche Notebook et sélectionnez la session Notebook créée dans le coin supérieur droit.

3. Copiez le code suivant dans le nouvel onglet Notebook et cliquez sur .

   import pandas as pd
   from pyspark.sql import SparkSession
   from pyspark.sql.types import StructType, StructField, StringType, IntegerType, LongType
   
   # 1. Prepare a Pandas DataFrame.
   pdf = pd.DataFrame({
       "id": ["1006"],
       "name": ["sl"]
   })
   
   # 2. Convert to a PySpark DataFrame.
   # (Optional: Explicitly define the schema to ensure correct data types)
   schema = StructType([
       StructField("id", StringType(), True),
       StructField("name", StringType(), True)
   ])
   
   df = spark.createDataFrame(pdf, schema=schema)
   
   # Write to Hologres.
   df.write \
     .format("hologres") \
     .option("username", "LTAI******") \
     .option("password", "mXYV******") \
     .option("jdbcurl", "jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/testdb") \
     .option("table", "test") \
     .mode("append") \
     .save()
   
   # Read data.
   readDf = spark.read\
     .format("hologres") \
     .option("username", "LTAI******") \
     .option("password", "mXYV******") \
     .option("jdbcurl", "jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/testdb") \
     .option("table", "test") \
     .load()
   
   readDf.select("id", "name").show(10)

   Le tableau suivant décrit les paramètres.

   Paramètre	Exemple	Description
   username	LTAI******	AccessKey ID de votre compte Alibaba Cloud. Nous vous recommandons d'utiliser la gestion des secrets pour traiter les informations sensibles. Pour plus de détails, consultez Gérer les informations sensibles à l'aide de la gestion des secrets.
   password	mXYV******	AccessKey Secret de votre compte Alibaba Cloud. Nous vous recommandons d'utiliser la gestion des secrets pour traiter les informations sensibles. Pour plus de détails, consultez Gérer les informations sensibles à l'aide de la gestion des secrets.
   jdbcurl	jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/testdb	URL de connexion JDBC de l'instance Hologres.

4. Vérifiez le résultat.