:Use Spark to import data

Informações básicas

O Spark é um mecanismo de análise que processa grandes volumes de dados de forma centralizada. O Hologres integra-se ao Apache Spark e ao E-MapReduce (EMR) Spark para ajudar você a construir data warehouses com eficiência. O Hologres fornece o conector Spark, que permite gravar dados no Hologres em modo batch. Com o Spark, é possível ler dados de diversas fontes, como arquivos, tabelas Hive, tabelas MySQL e tabelas PostgreSQL.

O Hologres é compatível com PostgreSQL. Use o Spark para ler dados do Hologres com base no PostgreSQL. Em seguida, extraia, transforme e carregue (ETL) os dados e grave os dados processados de volta no Hologres ou em outros destinos.

Pré-requisitos

• A versão da sua instância do Hologres é V0.9 ou posterior. Visualize a versão da sua instância do Hologres na página de detalhes da instância no console do Hologres. Se a versão da sua instância do Hologres for anterior à V0.9, atualize manualmente sua instância do Hologres no console do Hologres ou entre no grupo DingTalk do Hologres para obter suporte técnico. Para mais informações sobre como atualizar manualmente sua instância do Hologres no console do Hologres, consulte Atualizações de instância. Para mais informações sobre como obter suporte técnico, consulte Obter suporte online para o Hologres.

• Um ambiente Spark com versão suportada pelo Hologres está instalado. Isso permite executar o comando spark-shell no ambiente Spark.

Uso de conexões

Ao usar o conector Spark do Hologres para ler ou gravar dados, o sistema utiliza conexões Java Database Connectivity (JDBC). Os seguintes fatores afetam o número de conexões necessárias:

• Número de tarefas paralelas do Spark: obtenha esse valor na interface do usuário do Spark durante a execução de um job.

• Modo de gravação de dados: ao usar o conector Spark para gravar dados no modo fixed copy, cada tarefa paralela utiliza uma conexão JDBC. Ao usar a instrução INSERT para gravar dados, o número de conexões JDBC usadas por cada tarefa paralela corresponde ao valor de write_thread_size. Cada tarefa paralela de leitura de dados usa uma conexão JDBC.

• Outros: quando um job inicia, pode ser necessário obter informações de schema. Nesse caso, uma conexão JDBC pode ser usada por um curto período.

Calcule o número total de conexões JDBC usadas por um job com as seguintes fórmulas:

• Modo fixed copy: Parallelism × 1 + 1

• Instrução INSERT: Parallelism × write_thread_size + 1

(Recomendado) Usar o conector Spark para gravar dados no Hologres

Recomendamos o uso do conector Spark integrado do Hologres para gravar dados. O conector Spark funciona em conjunto com o Holo Client. Em comparação com outros métodos de gravação, o conector Spark oferece melhor desempenho. Para gravar dados com o conector Spark, siga as etapas abaixo. Para mais informações sobre o código de exemplo, consulte a seção Exemplo de uso do conector Spark para gravar dados no Hologres neste tópico.

Usar o Spark para ler dados de uma fonte específica e gravá-los no Hologres

1. Leia dados de uma fonte específica.

   O Spark permite ler dados de diferentes tipos de fontes. Os exemplos a seguir mostram como ler dados do Hologres ou de outro tipo de fonte:

   • Leitura de dados do Hologres

     O Hologres é compatível com PostgreSQL. Use o Spark para ler dados do Hologres com base no driver JDBC do PostgreSQL. O código de exemplo a seguir serve apenas como referência.

     Nota

     Antes de ler dados do Hologres, baixe o pacote PostgreSQL JDBC JAR no site oficial. Neste exemplo, usamos postgresql-42.2.18. Em seguida, execute o comando ./bin/spark-shell --jars /path/to/postgresql-42.2.18.jar no spark-shell para carregar o pacote PostgreSQL JDBC JAR. Também é possível carregar o pacote PostgreSQL JDBC JAR junto com o pacote JAR do conector Spark do Hologres.

     // Read from some table, for example: tb008
     val readDf = spark.read
       .format("jdbc") // Read Hologres data based on the PostgreSQL JDBC driver.
       .option("driver","org.postgresql.Driver")
       .option("url", "jdbc:postgresql://Ip:Por/test_database")
       .option("dbtable", "tb008")
       .option("user", "your_username")
       .option("password", "your_password")
       .load()

     O conector Spark V1.3.2 e versões posteriores permitem ler dados do Hologres e otimizar o desempenho de leituras paralelas. Comparado ao driver JDBC do PostgreSQL, o conector Spark possibilita configurar o paralelismo de leitura e suporta sharding de dados no Hologres para implementar leituras paralelas, melhorando significativamente o desempenho. Código de exemplo:

     val spark = SparkSession
     .builder
     .appName("ReadFromHologres")
     .master("local[*]")
     .getOrCreate()
     
     spark.sparkContext.setLogLevel("WARN")
     
     import spark.implicits._
     
     val schema = StructType(Array(
       StructField("id", LongType),
       StructField("counts", IntegerType),
       StructField("name", StringType, false),
       StructField("price", DecimalType(38, 12)),
       StructField("out_of_stock", BooleanType)
     ))
     
     val readDf = spark.read
     .format("hologres")
     .schema(schema) // Optional. If you do not specify a schema, all fields in the Hologres table are read by default.
     .option("username", "your_username")
     .option("password", "your_password")
     .option("jdbcurl", "jdbc:postgresql://hologres_endpoint/test_db")
     .option("table", "tb008")
     .option("scan_parallelism", "10") // The default parallelism for reading data from Hologres. The maximum value is the shard count of the Hologres table.
     .load()

   • Leitura de dados de outra fonte, como um arquivo Parquet

     Use o Spark para ler dados de outras fontes, como arquivos Parquet ou tabelas Hive. O exemplo a seguir fornece o código de amostra:

     import org.apache.spark.{SparkConf, SparkContext}
     import org.apache.spark.sql.hive.HiveContext
     
     val sparkConf = new SparkConf()
     val sc = new SparkContext(sparkConf)
     val hiveContext = new HiveContext(sc)
     
     // Read from some table, for example: phone
     val readDf = hiveContext.sql("select * from hive_database.phone")

2. Grave os dados no Hologres.

   import com.alibaba.hologres.spark2.sink.SourceProvider
   
   -- Write to hologres table
   df.write
     .format("hologres")
     .option(SourceProvider.USERNAME, "your_username")
     .option(SourceProvider.PASSWORD, "your_password")
     .option(SourceProvider.ENDPOINT, "Ip:Port")
     .option(SourceProvider.DATABASE, "test_database")
     .option(SourceProvider.TABLE, table)
     .option(SourceProvider.WRITE_BATCH_SIZE, 512) -- The maximum number of requests allowed in a batch.
     .option(SourceProvider.INPUT_DATA_SCHEMA_DDL, df.schema.toDDL) -- This parameter is required only for Spark 3.X.
     .mode(SaveMode.Append) // This parameter is required only for Spark 3.X.
     .save()

Usar o Spark para gravar dados no Hologres em tempo real

1. Execute a seguinte instrução para criar a tabela de destino no Hologres:

   CREATE TABLE test_table_stream
   (
       value text,
       count bigint
   );

2. Leia dados da sua máquina local. Colete estatísticas de frequência de palavras e grave-as no Hologres em tempo real. O exemplo a seguir fornece o código de amostra:

   • Código

      val spark = SparkSession
           .builder
           .appName("StreamToHologres")
           .master("local[*]")
           .getOrCreate()
     
         spark.sparkContext.setLogLevel("WARN")
         import spark.implicits._
     
         val lines = spark.readStream
           .format("socket")
           .option("host", "localhost")
           .option("port", 9999)
           .load()
     
         -- Split the lines into words
         val words = lines.as[String].flatMap(_.split(" "))
     
         -- Generate running word count
         val wordCounts = words.groupBy("value").count()
     
         wordCounts.writeStream
             .outputMode(OutputMode.Complete())
             .format("hologres")
             .option(SourceProvider.USERNAME, "your_username")
             .option(SourceProvider.PASSWORD, "your_password")
             .option(SourceProvider.JDBCURL, "jdbc:postgresql://Ip:Port/dbname")
             .option(SourceProvider.TABLE, "test_table_stream")
             .option("batchsize", 1)
             .option("isolationLevel", "NONE")
             .option("checkpointLocation", checkpointLocation)
             .start()
             .awaitTermination()

   • Parâmetros

     Parâmetro	Valor padrão	Obrigatório	Descrição
     username	Sem valor padrão	Sim	O AccessKey ID da sua conta Alibaba Cloud. Obtenha o AccessKey ID na página Security Management. Recomendamos configurar variáveis de ambiente e obter o AccessKey ID e o AccessKey secret a partir delas. Isso ajuda a reduzir o risco de vazamento.
     password	Sem valor padrão	Sim	O AccessKey secret da sua conta Alibaba Cloud. Obtenha o AccessKey secret na página Security Management. Recomendamos configurar variáveis de ambiente e obter o AccessKey ID e o AccessKey secret a partir delas. Isso ajuda a reduzir o risco de vazamento.
     table	Sem valor padrão	Sim	Nome da tabela do Hologres onde os dados serão gravados.
     endpoint	Sem valor padrão	Configure este parâmetro ou o parâmetro JDBCURL.	O endpoint da sua instância do Hologres. Obtenha o endpoint da sua instância do Hologres na seção Network Information da página de detalhes da instância no console do Hologres.
     database	Sem valor padrão	Configure este parâmetro ou o parâmetro JDBCURL.	Nome do banco de dados do Hologres onde a tabela de destino reside.
     jdbcurl	Sem valor padrão	Configure apenas este parâmetro ou os parâmetros ENDPOINT e DATABASE.	A JDBC URL da sua instância do Hologres.
     copy_write_mode	true	Não	Define se os dados serão gravados no modo fixed copy. O fixed copy é um novo recurso suportado no Hologres V1.3. Nesse modo, os dados são gravados em streaming em vez de batches. Portanto, a gravação no modo fixed copy oferece maior throughput, menor latência de dados e consome menos recursos de memória do cliente do que a gravação via instrução INSERT. Nota Para usar o modo fixed copy, a versão do conector deve ser V1.3.0 ou posterior e a versão do mecanismo do Hologres deve ser V1.3.34 ou posterior.
     copy_write_format	false	Não	Define se dados incorretos (dirty data) devem ser verificados. Este parâmetro só tem efeito quando COPY_WRITE_MODE está definido como true. Se definido como true e houver geração de dados incorretos, o sistema consegue localizar a linha cuja gravação falhou. Nota A verificação de dados incorretos afeta negativamente o desempenho de gravação. Recomendamos definir este parâmetro como true apenas durante processos de troubleshooting.
     bulk_load	true	Não	Define se os dados serão gravados no Hologres no modo batch copy. No modo fixed copy, os dados são gravados em streaming. Nota No Hologres V2.1, o desempenho de gravação em tabelas sem chaves primárias foi otimizado. Nessa versão, a operação de gravação em lote numa tabela sem chave primária não adquire um bloqueio de tabela, mas sim um bloqueio de linha. Assim, gravações em lote podem ocorrer em paralelo com operações realizadas via fixed plans, aumentando a eficiência e o paralelismo no processamento de dados. Se a versão do conector for V1.4.0 ou posterior, a versão do mecanismo do Hologres deve ser V2.1.0 ou posterior.
     max_cell_buffer_size	20971520 (20 MB)	Não	Comprimento máximo de um campo quando COPY_WRITE_MODE está definido como true.
     copy_write_dirty_data_check	false	Não	Define se dados incorretos devem ser verificados. Se definido como true e houver geração de dados incorretos, o sistema consegue localizar a linha cuja gravação falhou. A verificação de dados incorretos afeta negativamente o desempenho de gravação. Recomendamos definir este parâmetro como true apenas durante processos de troubleshooting. Nota Este parâmetro só tem efeito quando COPY_WRITE_MODE está definido como true.
     copy_write_direct_connect	true para cenários que permitem conexões diretas	Não	Este parâmetro só tem efeito quando COPY_WRITE_MODE está definido como true. A quantidade de dados graváveis no modo copy depende do throughput do endpoint VPC. O sistema verifica se o ambiente pode se conectar diretamente ao nó FE do Hologres durante a gravação no modo copy. Conexões diretas são usadas por padrão se o ambiente puder se conectar diretamente ao nó FE do Hologres. Se este parâmetro for definido como false, conexões diretas não serão usadas.
     input_data_schema_ddl	Sem valor padrão	Parâmetro obrigatório para Spark 3.X. Configure o valor no formato <your_DataFrame>.schema.toDDL.	Instrução DDL para DataFrame no Spark.
     write_mode	INSERT_OR_REPLACE	Não	Política usada para lidar com conflitos de chave primária. Obrigatório se a tabela de destino tiver chave primária. Valores válidos: INSERT_OR_IGNORE: descarta os dados a serem gravados se ocorrer conflito de chave primária. INSERT_OR_UPDATE: atualiza as colunas relevantes na tabela de destino se ocorrer conflito de chave primária. INSERT_OR_REPLACE: atualiza todas as colunas na tabela de destino se ocorrer conflito de chave primária.
     write_batch_size	512	Não	Número máximo de requisições permitidas em um lote dentro de uma thread para gravação de dados. Se o total de requisições PUT atingir o limite especificado pelo parâmetro write_batch_size após o tratamento de conflitos conforme o parâmetro write_mode, os dados são enviados em lote.
     write_batch_byte_size	2 MB	Não	Número máximo de bytes permitidos em um lote dentro de uma thread para gravação de dados. Se o total de bytes das requisições PUT atingir o limite especificado pelo parâmetro WRITE_BATCH_BYTE_SIZE após o tratamento de conflitos conforme o parâmetro WRITE_MODE, os dados são enviados em lote.
     write_max_interval_ms	10000 ms	Não	Intervalo de tempo para envio de dados em lote.
     write_fail_strategy	TYR_ONE_BY_ONE	Não	Política usada para lidar com falhas de envio. Se um lote não puder ser enviado, o Holo Client envia as entradas de dados uma a uma, na sequência especificada. Caso uma entrada específica falhe, o Holo Client retorna uma mensagem de erro contendo informações sobre essa entrada.
     write_thread_size	1	Não	Número de threads paralelas usadas para gravar dados. Cada thread ocupa uma conexão. O número total de conexões ocupadas por um job Spark varia conforme o paralelismo do Spark. Calcule o total de conexões usando a fórmula: Total de conexões = Valor de spark.default.parallelism × Valor de WRITE_THREAD_SIZE.
     dynamic_partition	false	Não	Define se uma partição será criada automaticamente caso os dados sejam gravados em uma tabela pai sem partições. Valores válidos: true e false. true: cria automaticamente uma partição se os dados forem gravados em uma tabela pai sem partições.
     retry_count	3	Não	Número máximo de tentativas permitidas para gravar e consultar dados em caso de falha de conexão.
     retry_sleep_init_ms	1000 ms	Não	O tempo consumido pelas novas tentativas de uma requisição é calculado pela fórmula: retry_sleep_init_ms + retry_count × retry_sleep_step_ms.
     retry_sleep_step_ms	10*1000 ms	Não	O tempo consumido pelas novas tentativas de uma requisição é calculado pela fórmula: retry_sleep_init_ms + retry_count × retry_sleep_step_ms.
     connection_max_idle_ms	60000 ms	Não	Tempo limite de ociosidade para conexões usadas na leitura e gravação de dados. Se uma conexão permanecer ociosa por um período superior ao tempo limite especificado, o Holo Client libera a conexão automaticamente.
     fixed_connection_mode	false	Não	Define se conexões fixas serão usadas. Em modos que não são fixed copy, como o modo INSERT, gravações de dados e consultas pontuais não ocupam conexões. Nota O recurso de conexão fixa está em versão beta e está disponível somente quando a versão do conector é V1.2.0 ou posterior e a versão do mecanismo do Hologres é V1.3.0 ou posterior.
     scan_batch_size	256	Não	Número de linhas verificadas a cada vez durante a leitura de dados do Hologres.
     scan_timeout_seconds	60	Não	Tempo limite da operação de varredura ao ler dados do Hologres. Unidade: segundos.
     scan_parallelism	10	Não	Número de shards usados para ler dados do Hologres. O valor máximo corresponde à contagem de shards da tabela do Hologres. Durante a execução do job, os shards são alocados às tarefas do Spark para leitura de dados.

Mapeamentos de tipos de dados

A tabela a seguir descreve os mapeamentos de tipos de dados entre Spark e Hologres.