E-MapReduce:Leitura e gravação no Hologres

Exemplo 1: Sessão SQL

Leia e grave dados no Hologres usando uma sessão SQL.

1. Crie uma sessão SQL. Para obter detalhes, consulte Gerenciar sessões SQL.

   Ao criar a sessão, selecione a conexão de rede criada na etapa anterior na lista de conexões de rede. Na seção Spark Configuration, adicione os seguintes parâmetros para carregar o 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

   A tabela a seguir descreve os parâmetros.

   Parâmetro	Exemplo	Descrição
   spark.emr.serverless.user.defined.jars	oss://<bucket>/hologres-connector-spark-3.x-<version>.jar	Caminho para o arquivo JAR definido pelo usuário.
   spark.sql.catalog.hologres_external_test_db	com.alibaba.hologres.spark3.HoloTableCatalog	Configura uma fonte de dados do Hologres como um catálogo externo no Spark 3.x. Valor fixo.
   spark.sql.catalog.hologres_external_test_db.username	LTAI******	AccessKey ID da sua conta Alibaba Cloud. Recomendamos usar o gerenciamento de segredos para lidar com informações sensíveis. Para obter detalhes, consulte Gerenciar informações sensíveis usando o gerenciamento de segredos.
   spark.sql.catalog.hologres_external_test_db.password	mXYV******	AccessKey Secret da sua conta Alibaba Cloud. Recomendamos usar o gerenciamento de segredos para lidar com informações sensíveis. Para obter detalhes, consulte Gerenciar informações sensíveis usando o gerenciamento de segredos.
   spark.sql.catalog.hologres_external_test_db.jdbcurl	jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/testdb	JDBC connection URL da instância do Hologres.

   É possível personalizar a parte hologres_external_test_db nos nomes dos parâmetros.

2. Na página Data Development, crie um job SparkSQL e selecione a sessão SQL criada no canto superior direito.

   Para mais informações, consulte Desenvolvimento SparkSQL.

3. Copie o código abaixo para a nova aba SparkSQL e clique em 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;

   

Exemplo 2: Job de streaming

Este exemplo em PySpark lê dados do Kafka e os grava no Hologres como um job de streaming.

1. Neste exemplo de código, substitua as informações do Kafka e a tabela do Hologres pelos seus valores reais.

   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. Faça upload do arquivo.

   1. Na página Artifacts, clique em Upload File.

   2. Na caixa de diálogo Upload File, clique na área de upload para selecionar o arquivo Python da etapa anterior ou arraste o arquivo para a área de upload.

3. Crie e execute o job de streaming.

   1. Na página Development, clique no ícone (Create).

   2. Na caixa de diálogo exibida, insira um Name, selecione PySpark na lista Application (Streaming) e clique em OK.

   3. Na nova aba de desenvolvimento, configure os parâmetros a seguir e mantenha os demais com as configurações padrão. Em seguida, clique em Publish.

      Parâmetro	Descrição
      Main Python Resources	Selecione o arquivo Python enviado na etapa anterior.
      Engine Version	Selecione uma versão compatível do Spark. Este exemplo usa esr-4.6.0.
      Network Connection	Selecione a conexão de rede criada na Etapa 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 Para obter uma descrição detalhada dos parâmetros, consulte Exemplo 1: Sessão SQL.

   4. Após publicar o job, clique em Go to O&M. Na página que se abre, clique em Start.

4. Verifique o resultado.

   1. Envie mensagens para o Kafka.

      

   2. Consulte os dados usando Spark SQL.

Exemplo 3: Sessão Notebook

1. Crie uma sessão Notebook. Para obter detalhes, consulte Gerenciar sessões Notebook.

   Ao criar a sessão, selecione a conexão de rede criada na etapa anterior na lista de conexões de rede. Na seção Spark Configuration, adicione o seguinte parâmetro para carregar o 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. Na página Data Development, crie um job Notebook e selecione a sessão Notebook criada no canto superior direito.

3. Copie o código abaixo para a nova aba Notebook e clique em .

   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)

   A tabela a seguir descreve os parâmetros.

   Parâmetro	Exemplo	Descrição
   username	LTAI******	AccessKey ID da sua conta Alibaba Cloud. Recomendamos usar o gerenciamento de segredos para lidar com informações sensíveis. Para obter detalhes, consulte Gerenciar informações sensíveis usando o gerenciamento de segredos.
   password	mXYV******	AccessKey Secret da sua conta Alibaba Cloud. Recomendamos usar o gerenciamento de segredos para lidar com informações sensíveis. Para obter detalhes, consulte Gerenciar informações sensíveis usando o gerenciamento de segredos.
   jdbcurl	jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/testdb	JDBC connection URL da instância do Hologres.

4. Verifique o resultado.