Hologres:Read and write Hologres with Spark

spark-sql --jars hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar \
--conf spark.sql.catalog.hologres_external_test_db=com.alibaba.hologres.spark3.HoloTableCatalog \
--conf spark.sql.catalog.hologres_external_test_db.username=*** \
--conf spark.sql.catalog.hologres_external_test_db.password=*** \
--conf spark.sql.catalog.hologres_external_test_db.jdbcurl=jdbc:postgresql://hgpostcn-cn-***-vpc-st.hologres.aliyuncs.com:80/test_db

Para executar jobs Spark em diferentes linguagens de programação:

• Scala

  • Crie um arquivo sparkselect.scala com o código de exemplo e execute o job com o comando a seguir.

    -- Load dependencies.
    spark-shell --jars hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar
    
    -- For local testing, use an absolute path to load the file.
    scala> :load D:/sparkselect.scala

  • Alternativamente, após carregar as dependências, cole o código de exemplo diretamente no shell para executá-lo.

• Java

  Utilize uma ferramenta de desenvolvimento para importar o código de exemplo e empacotá-lo usando Maven. Por exemplo, se o JAR empacotado for nomeado spark_select.jar, execute o job com o comando a seguir.

  -- Use an absolute path for the job JAR.
  spark-submit --class SparkSelect --jars hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar  D:\spark_select.jar

• Python

  Após executar o comando abaixo, cole o código de exemplo diretamente no shell para executá-lo.

  pyspark --jars hologres-connector-spark-3.x-1.5.2-jar-with-dependencies.jar

Parâmetros

Parâmetros gerais

Parameter	Default	Required	Description
username	None	Yes	O AccessKey ID da sua conta Alibaba Cloud. Obtenha-o na página AccessKey Management. Alternativamente, o nome da conta de usuário. Após criar uma conta, conceda a ela as permissões necessárias no banco de dados. Para mais informações, consulte Modelo de permissões do Hologres. Em seguida, conecte-se ao HoloWeb e execute consultas para verificar as permissões.
password	None	Yes	O AccessKey Secret correspondente ao seu AccessKey ID. Para mais informações, consulte Criar um AccessKey. Alternativamente, a senha da conta.
table	None	Yes	O nome da tabela Hologres para leitura ou gravação. Nota Ao ler dados, utilize o parâmetro read.query como alternativa.
jdbcurl	None	Yes	A URL JDBC da API de dados em tempo real do Hologres, no formato jdbc:postgresql://<host>:<port>/<db_name>. Para encontrar esses detalhes, acesse o console do Hologres, clique em Instances no painel de navegação à esquerda, selecione sua instância alvo e encontre o host e o número da porta na seção Network Information da página Instance Details.
enable_serverless_computing	false	No	Define se deve ser usado um recurso de computação serverless. Este parâmetro aplica-se apenas a operações de leitura e gravação no modo bulk_load. Para mais informações, consulte Guia do usuário do Serverless Computing.
serverless_computing_query_priority	3	No	A prioridade de execução para Serverless Computing.
statement_timeout_seconds	28800 (8 hours)	No	A duração do tempo limite de execução da consulta, em segundos.
retry_count	3	No	O número de tentativas em caso de falha na conexão.
direct_connect	Uma conexão direta é usada por padrão, se suportada.	No	Define se a conexão deve ser feita diretamente ao FrontEnd do Hologres (nó de acesso). O throughput de rede do endpoint costuma ser o gargalo para operações de dados em lote. Por padrão, o conector usa uma conexão direta, se disponível, para melhorar o throughput. Defina como false para desativar esse comportamento.

Parâmetros de gravação

O conector Hologres suporta o parâmetro SaveMode do Spark. Para SQL, isso corresponde a INSERT INTO ou INSERT OVERWRITE. Para um DataFrame, defina o SaveMode como Append ou Overwrite ao gravar dados. O modo Overwrite cria uma tabela temporária para a operação de gravação e substitui a tabela original após o sucesso. Utilize este modo apenas quando necessário.

Parameter	Previous name	Default	Required	Description
write.mode	copy_write_mode	auto	No	O modo de gravação. Para uma comparação dos modos de gravação, consulte Modos de gravação em lote. Valores válidos: auto (Padrão): O conector seleciona automaticamente o modo ideal com base na versão do Hologres e nos metadados da tabela de destino. A lógica de seleção é a seguinte: Se a instância Hologres for v2.2.25 ou posterior e a tabela tiver uma chave primária, o modo bulk_load_on_conflict será utilizado. Se a instância Hologres for v2.1.0 ou posterior e a tabela não tiver uma chave primária, o modo bulk_load será utilizado. Se a instância Hologres for v1.3 ou posterior, o modo stream será utilizado. Nos demais casos, o modo insert será utilizado. stream: Utiliza Fixed Plan para acelerar a execução de SQL. Em um Fixed Plan, COPY é um recurso introduzido no Hologres v1.3. Comparado ao método INSERT, o COPY oferece maior throughput (devido à sua natureza de streaming), menor latência de dados e consumo reduzido de memória do cliente (pois não faz batch de dados). Nota Requer conector Hologres v1.3.0 ou posterior e Hologres v1.3.34 ou posterior. bulk_load: Utiliza COPY em lote. Comparado ao COPY via streaming em um Fixed Plan, o COPY em lote impõe uma carga menor à instância Hologres sob condições de alto RPS, mas suporta apenas gravação em tabelas sem chave primária. Nota Requer conector Hologres v1.4.2 ou posterior e Hologres v2.1.0 ou posterior. bulk_load_on_conflict: Utiliza COPY em lote para gravar em tabelas com chave primária e consegue lidar com chaves primárias duplicadas. Por padrão, a importação de dados em lote para uma tabela Hologres com chave primária aciona um bloqueio de tabela, limitando operações de gravação simultâneas de múltiplas conexões. O conector pode redistribuir os dados pela chave de distribuição da tabela de destino, permitindo que cada tarefa Spark grave em um único shard. Isso reduz o bloqueio de tabela para um bloqueio no nível de shard, habilitando gravações concorrentes e melhorando o desempenho de gravação. Como cada conexão mantém dados de apenas alguns shards, essa otimização também reduz significativamente o número de arquivos pequenos e diminui o uso de memória do Hologres. Testes mostram que reparticionar dados antes de uma gravação concorrente pode reduzir a carga do sistema em cerca de 67% em comparação ao modo stream. Nota Requer conector Hologres v1.4.2 ou posterior e Hologres v2.2.25 ou posterior. insert: Grava dados usando o método INSERT. stage: Usa uma tabela Stage para gravações em lote. Este modo importa dados em tempo quase real através de um estágio de armazenamento temporário e depois carrega os dados na tabela de destino usando uma instrução INSERT FROM SELECT. Em cenários de lote, este modo oferece melhor desempenho e menor carga no sistema do que outros modos. Ele também suporta o uso de recursos de computação serverless. Tanto operações INSERT INTO quanto INSERT OVERWRITE são suportadas. Requer Hologres v4.1.0 ou posterior e conector v1.6.1 ou posterior.
write.copy.max_buffer_size	max_cell_buffer_size	52428800 (50 MB)	No	O tamanho máximo do buffer local ao gravar no modo COPY. Geralmente não é necessário ajustar esse valor. No entanto, aumente-o se ocorrer um estouro de buffer ao gravar campos grandes, como strings extremamente longas.
write.copy.dirty_data_check	copy_write_dirty_data_check	false	No	Define se deve verificar dados incorretos (dirty data). Se ativado, esse recurso identifica exatamente a linha que falhou na gravação. Contudo, isso afeta o desempenho de gravação. Mantenha este recurso desativado, exceto para solução de problemas.
write.on_conflict_action	INSERT_OR_REPLACE	INSERT_OR_REPLACE	No	A ação a ser tomada quando uma operação de gravação encontra um conflito de chave primária na tabela de destino. INSERT_OR_IGNORE: Se ocorrer um conflito de chave primária, os dados não serão gravados. INSERT_OR_UPDATE: Se ocorrer um conflito de chave primária, as colunas especificadas serão atualizadas. INSERT_OR_REPLACE: Se ocorrer um conflito de chave primária, todas as colunas serão atualizadas.
write.stage.compression	None	false	No	Entra em vigor quando write.mode está definido como stage. Definir isso como true ativa a compressão durante gravações stage. Requer Hologres v4.2.8 ou posterior e conector v1.6.2 ou posterior.

Os parâmetros a seguir entram em vigor apenas quando write.mode está definido como insert.

Parameter	Previous name	Default	Required	Description
write.insert.dynamic_partition	dynamic_partition	false	No	Quando write.mode é insert, definir isto como true cria automaticamente partições inexistentes ao gravar em uma tabela pai particionada.
write.insert.batch_size	write_batch_size	512	No	O tamanho máximo de lote para cada thread de gravação. Um commit em lote é acionado quando o número de operações Put atinge esse valor.
write.insert.batch_byte_size	write_batch_byte_size	2097152 (2 MB)	No	O tamanho máximo de lote em bytes para cada thread de gravação. O valor padrão é 2 MB. Um commit em lote é acionado quando o tamanho em bytes dos dados Put atinge esse valor.
write.insert.max_interval_ms	write_max_interval_ms	10000	No	Um commit em lote é acionado se o tempo decorrido desde o último commit exceder este valor.
write.insert.thread_size	write_thread_size	1	No	O número de threads de gravação concorrentes. Cada thread usa uma conexão de banco de dados.
write.rps_limit	None	-1	No	O limite de taxa para gravações por tarefa, em linhas por segundo (RPS). O valor padrão -1 indica nenhum limite.

Parâmetros de leitura

Parameter	Previous name (v1.5.0 and earlier)	Default	Required	Description
read.mode	bulk_read	auto	No	O modo de leitura. Valores válidos: auto (Padrão): O conector seleciona automaticamente o modo ideal com base na versão do Hologres e nos metadados da tabela. A lógica de seleção é a seguinte: Se os campos a serem lidos incluírem o tipo de dados JSONB, o modo select será utilizado. Se a instância for v3.0.24 ou posterior, o modo bulk_read_compressed será utilizado. Nos demais casos, o modo bulk_read será utilizado. bulk_read: Utiliza o método COPY OUT para ler dados no formato arrow, que é várias vezes mais rápido que o modo select. Atualmente, não há suporte para leitura do tipo JSONB do Hologres. bulk_read_compressed: Lê dados compactados no formato Arrow usando o método COPY OUT, economizando aproximadamente 45% de largura de banda em comparação à leitura de dados não compactados. select: Utiliza instruções SELECT padrão para ler dados.
read.max_task_count	max_partition_count	80	No	Especifica o número máximo de tarefas concorrentes para leitura de dados. O conector divide a tabela em múltiplas partições, e cada partição é processada por uma única tarefa Spark. Se a contagem de shards da tabela for menor que este valor, o número de partições será limitado à contagem de shards.
read.copy.max_buffer_size	/	52428800 (50 MB)	No	O tamanho máximo do buffer local ao ler no modo COPY. Se ocorrer uma exceção devido a um campo grande, aumente este valor.
read.push_down_predicate	push_down_predicate	true	No	Define se o pushdown de predicado deve ser ativado. Quando ativado, operações como condições de filtro e poda de colunas são transferidas para a fonte de dados.
read.push_down_limit	push_down_limit	true	No	Define se o pushdown de LIMIT deve ser ativado.
read.select.batch_size	scan_batch_size	256	No	Entra em vigor quando read.mode está definido como select. Especifica o número de linhas a serem buscadas em uma única operação de varredura ao ler do Hologres.
read.select.timeout_seconds	scan_timeout_seconds	60	No	Entra em vigor quando read.mode está definido como select. Especifica a duração do tempo limite para uma operação de varredura ao ler do Hologres.
read.query	query	None	No	Use a query fornecida para ler do Hologres. Este parâmetro não pode ser usado junto com o parâmetro table. Nota Ao ler dados usando o método query, apenas uma única tarefa pode ser utilizada. Não há suporte para pushdown de predicado. Ao ler dados usando o método table, a operação de leitura é dividida em múltiplas Tasks que rodam em paralelo com base no ShardCount da tabela Hologres.
read.split.strategy	None	auto	No	Define a estratégia para dividir os dados da tabela em múltiplas tarefas para leituras paralelas. Valores válidos: auto (padrão): O conector seleciona automaticamente a estratégia ideal. shard: Fragmenta os dados com base nos shards do Hologres, e cada tarefa lê uma parte dos shards. Adequado para tabelas com distribuição clara de shards. range: Fragmenta os dados com base no intervalo de valores de uma coluna especificada. Esta configuração requer os parâmetros read.split.column, read.split.lower_bound, read.split.upper_bound e read.split.num. partition fragmenta os dados com base nos valores da coluna de partição. Esta opção deve ser usada com os parâmetros read.split.column e read.split.num. Aplica-se a tabelas particionadas ou views particionadas. Nota Suportado no conector v1.6.1 e posteriores.
read.split.column	None	None	No	O nome da coluna de divisão. Este parâmetro é obrigatório quando read.split.strategy é range ou partition.
read.split.lower_bound	None	None	No	O limite inferior de divisão é obrigatório quando read.split.strategy está definido como range. Especifica o valor do limite inferior para a divisão por intervalo.
read.split.upper_bound	None	None	No	O limite superior de divisão especifica o limite máximo para um shard baseado em intervalo. Este parâmetro é obrigatório quando read.split.strategy está definido como range.
read.split.num	None	None	No	O número de divisões. Este parâmetro é obrigatório quando read.split.strategy é range ou partition. Especifica o número de tarefas nas quais os dados são divididos para leitura.

Mapeamento de tipos de dados

Spark type	Hologres type
ShortType	SMALLINT
IntegerType	INT
LongType	BIGINT
StringType	TEXT
StringType	JSON
StringType	JSONB
DecimalType	NUMERIC(38, 18)
BooleanType	BOOL
DoubleType	DOUBLE PRECISION
FloatType	FLOAT
TimestampType	TIMESTAMPTZ
DateType	DATE
BinaryType	BYTEA
BinaryType	ROARINGBITMAP
ArrayType(IntegerType)	INT4[]
ArrayType(LongType)	INT8[]
ArrayType(FloatType)	FLOAT4[]
ArrayType(DoubleType)	FLOAT8[]
ArrayType(BooleanType)	BOOLEAN[]
ArrayType(StringType)	TEXT[]

Cálculo da contagem de conexões

O Hologres-Connector-Spark utiliza um determinado número de conexões JDBC para operações de leitura e gravação. A contagem de conexões depende dos seguintes fatores:

• Paralelismo do Spark, ou seja, o número de tarefas concorrentes executadas durante o job, visíveis na UI do Spark.

• O número de conexões usadas por tarefa:

  • Ao gravar no modo COPY, cada tarefa usa uma conexão JDBC.

  • Ao gravar no modo INSERT, cada tarefa usa write_thread_size conexões JDBC.

  • Ao ler dados, cada tarefa usa uma conexão JDBC.

• Outras operações: Um job pode usar brevemente uma conexão para tarefas como obtenção de schema ao iniciar.

Calcule a contagem total de conexões para um job com as seguintes fórmulas:

Item	Connections
Consulta de metadados do catálogo	1
Leitura de dados	parallelism * 1 + 1
Gravação no modo COPY	parallelism * 1 + 1
Gravação no modo INSERT	parallelism * write_thread_size + 1

Este cálculo pressupõe que a capacidade do Spark para tarefas concorrentes excede o número de tarefas geradas pelo job.

A capacidade do Spark para tarefas concorrentes depende de parâmetros configurados pelo usuário, como spark.executor.instances, e da política de divisão de blocos de arquivos do Hadoop. Para mais informações, consulte Apache Hadoop.

Notas de versão do Conector Spark

Version	Release date	New features	Bug fixes
1.6.1	2026-02	Adicionado o modo de gravação stage, que usa uma tabela stage temporária para melhorar o desempenho de gravação. Este recurso requer Hologres v4.1.0 ou posterior. Adicionado suporte para operações INSERT INTO e INSERT OVERWRITE no modo stage. Otimizada a lógica de limpeza de tabelas temporárias para operações overwrite. Adicionado o parâmetro write.rps_limit para controlar a taxa de gravação por tarefa.	Corrigido um problema de replay síncrono forçado que ocorria ao executar DDL durante uma operação overwrite.
1.6.0	2025-12	Adicionado o parâmetro read.split.strategy para suportar três estratégias de divisão de leitura: shard, range e partition. Adicionado suporte para o tipo de dados JSONB em leituras baseadas em Arrow. Adicionado o parâmetro write.insert.ignore_null_when_update para ignorar valores NULL durante operações UPDATE. Adicionado suporte para relatórios de cobertura de testes JaCoCo.	Corrigido um erro ao gravar valores date anteriores a 1970.
1.5.6	2025-11	Adicionado suporte para o método de autenticação AKV4. Adicionado o parâmetro write.copy.disable_right_join para otimizar o desempenho de gravação COPY.	Otimizada a lógica de verificação de tipos de colunas para gravações via catálogo.
1.5.5	2025-10	Adicionado suporte para gravação de dados de baixa precisão em tipos de dados de alta precisão. Adicionados prefixos como appname e taskid aos logs para simplificar a solução de problemas. Ativada a compressão Arrow LZ4 para operações COPY OUT.	Otimizada a formatação e validação de parâmetros.
1.5.4	2025-09	Adicionado o parâmetro remove_u0000 para remover automaticamente caracteres U+0000 de colunas de texto. Refatorado o Catálogo Hologres para mapear namespaces para schemas. Adicionada a classe utilitária RepartitionUtil para fragmentação de dados.	Documentação refatorada e links atualizados.
1.5.2	2025-08	Introduzido o Catálogo Hologres, que permite leitura e gravação no Hologres usando tabelas externas. Adicionado suporte para pushdown de predicado e pushdown de limite. Ativado o modo em lote bulk_read para operações de leitura, melhorando o desempenho em uma ordem de magnitude.	Desempenho de leitura otimizado.
1.5.0	2025-06	Adicionado suporte para pushdown de predicado, pushdown de limite e poda de colunas. Ativada a leitura de dados usando consultas SELECT. Ativado o modo em lote para operações de leitura.	Corrigidos problemas relacionados à leitura.
1.4.2	2025-04	Introduzido o modo de gravação bulk_load para gravação em lote em tabelas sem chave primária. Adicionado o modo bulk_load_on_conflict para lidar com conflitos de chave primária em tabelas com chave primária.	Desempenho de gravação otimizado.
1.3.2	2025-02	Adicionado suporte para leitura de dados do Hologres. Ativadas leituras paralelas por shard. Adicionado suporte para inferência automática de schema.	Estabilizada a funcionalidade inicial de leitura.
1.3.0	2025-01	Introduzido o modo de gravação stream (COPY fixo). Adicionado suporte para o recurso COPY fixo no Hologres v1.3.	Throughput e latência de gravação otimizados.