E-MapReduce:Read from and write to Hologres

Example 1: SQL session

This example shows how to read from and write to Hologres by using an SQL session.

1. Create an SQL session. For details, see Manage SQL Sessions.

   When you create the session, select the network connection that you created in the previous step from the network connection list. In the Spark Configuration section, add the following parameters to load the 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

   The following table describes the parameters.

   Parameter	Example	Description
   spark.emr.serverless.user.defined.jars	oss://<bucket>/hologres-connector-spark-3.x-<version>.jar	Specifies the path to the user-defined JAR file.
   spark.sql.catalog.hologres_external_test_db	com.alibaba.hologres.spark3.HoloTableCatalog	In Spark 3.x, this parameter is used to configure a Hologres data source as an external catalog. This is a fixed value.
   spark.sql.catalog.hologres_external_test_db.username	LTAI******	The AccessKey ID of your Alibaba Cloud account. We recommend that you use secrets management to handle sensitive information. For details, see Manage sensitive information by using secrets management.
   spark.sql.catalog.hologres_external_test_db.password	mXYV******	The AccessKey Secret of your Alibaba Cloud account. We recommend that you use secrets management to handle sensitive information. For details, see Manage sensitive information by using secrets management.
   spark.sql.catalog.hologres_external_test_db.jdbcurl	jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/testdb	The JDBC connection URL of the Hologres instance.

   You can customize the hologres_external_test_db part in the parameter names.

2. On the Data Development page, create a SparkSQL job and select the SQL session that you created from the upper-right corner.

   For more information, see SparkSQL development.

3. Copy the following code into the new SparkSQL tab and click 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;

   

Example 2: Streaming job

This PySpark example shows how to read data from Kafka and write it to Hologres as a streaming job.

1. In this code example, replace the Kafka information and Hologres table with your actual values.

   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. Upload the file.

   1. On the Artifacts page, click Upload File.

   2. In the Upload File dialog box, click the upload area to select the Python file from the previous step, or drag the file into the upload area.

3. Create and run the streaming job.

   1. On the Development page, click the (Create) icon.

   2. In the dialog box that appears, enter a Name, select PySpark from the Application (Streaming) list, and then click OK.

   3. In the new development tab, configure the following parameters and leave the others at their default settings. Then, click Publish.

      Parameter	Description
      Main Python Resources	Select the Python file that you uploaded in the previous step.
      Engine Version	Select a compatible Spark version. This example uses esr-4.6.0.
      Network Connection	Select the network connection that you created in Step 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 For a detailed description of the parameters, see Example 1: SQL session.

   4. After you publish the job, click Go to O&M. On the page that opens, click Start.

4. Verify the result.

   1. Send messages to Kafka.

      

   2. Query the data by using Spark SQL.

Example 3: Notebook session

1. Create a Notebook session. For details, see Manage Notebook Sessions.

   When you create the session, select the network connection that you created in the previous step from the network connection list. In the Spark Configuration section, add the following parameter to load the 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. On the Data Development page, create a Notebook job and select the Notebook session that you created from the upper-right corner.

3. Copy the following code into the new Notebook tab and click .

   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)

   The following table describes the parameters.

   Parameter	Example	Description
   username	LTAI******	The AccessKey ID of your Alibaba Cloud account. We recommend that you use secrets management to handle sensitive information. For details, see Manage sensitive information by using secrets management.
   password	mXYV******	The AccessKey Secret of your Alibaba Cloud account. We recommend that you use secrets management to handle sensitive information. For details, see Manage sensitive information by using secrets management.
   jdbcurl	jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/testdb	The JDBC connection URL of the Hologres instance.

4. Verify the result.