Como ingerir dados usando pg_azure_storage no Azure Cosmos DB para PostgreSQL

APLICA-SE A: Azure Cosmos DB para PostgreSQL (alimentado pela extensão de banco de dados Citus para PostgreSQL)

Este artigo mostra como usar a extensão pg_azure_storage PostgreSQL para manipular e carregar dados em seu Azure Cosmos DB para PostgreSQL diretamente do Armazenamento de Blob do Azure (ABS). O ABS é um serviço de armazenamento escalável, durável e seguro nativo da nuvem. Essas características o tornam uma boa escolha de armazenar e mover dados existentes para a nuvem.

Preparar o armazenamento de banco de dados e blob

Para carregar dados do Armazenamento de Blobs do Azure, instale a pg_azure_storage extensão PostgreSQL em seu banco de dados:

SELECT * FROM create_extension('azure_storage');

Preparamos um conjunto de dados de demonstração pública para este artigo. Para usar seu próprio conjunto de dados, siga migrar seus dados locais para o armazenamento em nuvem para saber como obter seus conjuntos de dados de forma eficiente no Armazenamento de Blobs do Azure.

Listar conteúdo do contêiner

Há uma conta de demonstração do Armazenamento de Blobs do Azure e um contêiner pré-criados para esse procedimento. O nome do contêiner é github, e está na pgquickstart conta. Podemos ver facilmente quais arquivos estão presentes no contêiner usando a azure_storage.blob_list(account, container) função.

SELECT path, bytes, pg_size_pretty(bytes), content_type
  FROM azure_storage.blob_list('pgquickstart','github');

-[ RECORD 1 ]--+-------------------
path           | events.csv.gz
bytes          | 41691786
pg_size_pretty | 40 MB
content_type   | application/x-gzip
-[ RECORD 2 ]--+-------------------
path           | users.csv.gz
bytes          | 5382831
pg_size_pretty | 5257 kB
content_type   | application/x-gzip

Você pode filtrar a saída usando uma cláusula SQL WHERE regular ou usando o prefix blob_list parâmetro do UDF. Este último filtra as linhas retornadas no lado do Armazenamento de Blobs do Azure.

SELECT * FROM azure_storage.blob_list('pgquickstart','github','e');

-[ RECORD 1 ]----+---------------------------------
path             | events.csv.gz
bytes            | 41691786
last_modified    | 2022-10-12 18:49:51+00
etag             | 0x8DAAC828B970928
content_type     | application/x-gzip
content_encoding |
content_hash     | 473b6ad25b7c88ff6e0a628889466aed

SELECT *
  FROM azure_storage.blob_list('pgquickstart','github')
 WHERE path LIKE 'e%';

-[ RECORD 1 ]----+---------------------------------
path             | events.csv.gz
bytes            | 41691786
last_modified    | 2022-10-12 18:49:51+00
etag             | 0x8DAAC828B970928
content_type     | application/x-gzip
content_encoding |
content_hash     | 473b6ad25b7c88ff6e0a628889466aed

Carregar dados do ABS

Carregar dados com o comando COPY

Comece criando um esquema de exemplo.

CREATE TABLE github_users
(
	user_id bigint,
	url text,
	login text,
	avatar_url text,
	gravatar_id text,
	display_login text
);

CREATE TABLE github_events
(
	event_id bigint,
	event_type text,
	event_public boolean,
	repo_id bigint,
	payload jsonb,
	repo jsonb,
	user_id bigint,
	org jsonb,
	created_at timestamp
);

CREATE INDEX event_type_index ON github_events (event_type);
CREATE INDEX payload_index ON github_events USING GIN (payload jsonb_path_ops);

SELECT create_distributed_table('github_users', 'user_id');
SELECT create_distributed_table('github_events', 'user_id');

Carregar dados nas tabelas torna-se tão simples quanto chamar o COPY comando.

-- download users and store in table

COPY github_users
FROM 'https://pgquickstart.blob.core.windows.net/github/users.csv.gz';

-- download events and store in table

COPY github_events
FROM 'https://pgquickstart.blob.core.windows.net/github/events.csv.gz';

Observe como a extensão reconheceu que as URLs fornecidas para o comando copy são do Armazenamento de Blobs do Azure, os arquivos que apontamos foram compactados por gzip e isso também foi tratado automaticamente para nós.

O COPY comando suporta mais parâmetros e formatos. No exemplo acima, o formato e a compactação foram selecionados automaticamente com base nas extensões de arquivo. No entanto, você pode fornecer o formato diretamente semelhante ao comando regular COPY .

COPY github_users
FROM 'https://pgquickstart.blob.core.windows.net/github/users.csv.gz'
WITH (FORMAT 'csv');

Atualmente, a extensão suporta os seguintes formatos de arquivo:

Carregar dados com blob_get()

O COPY comando é conveniente, mas limitado em flexibilidade. Internamente, COPY usa a blob_get função, que você pode usar diretamente para manipular dados em cenários mais complexos.

SELECT *
  FROM azure_storage.blob_get(
         'pgquickstart', 'github',
         'users.csv.gz', NULL::github_users
       )
 LIMIT 3;

-[ RECORD 1 ]-+--------------------------------------------
user_id       | 21
url           | https://api.github.com/users/technoweenie
login         | technoweenie
avatar_url    | https://avatars.githubusercontent.com/u/21?
gravatar_id   |
display_login | technoweenie
-[ RECORD 2 ]-+--------------------------------------------
user_id       | 22
url           | https://api.github.com/users/macournoyer
login         | macournoyer
avatar_url    | https://avatars.githubusercontent.com/u/22?
gravatar_id   |
display_login | macournoyer
-[ RECORD 3 ]-+--------------------------------------------
user_id       | 38
url           | https://api.github.com/users/atmos
login         | atmos
avatar_url    | https://avatars.githubusercontent.com/u/38?
gravatar_id   |
display_login | atmos

Com essa função, você pode manipular dados instantaneamente em consultas complexas e fazer importações como INSERT FROM SELECT.

INSERT INTO github_users
     SELECT user_id, url, UPPER(login), avatar_url, gravatar_id, display_login
       FROM azure_storage.blob_get('pgquickstart', 'github', 'users.csv.gz', NULL::github_users)
      WHERE gravatar_id IS NOT NULL;

INSERT 0 264308

No comando acima, filtramos os dados para contas com um gravatar_id presente e caixa superior seus logins na hora.

Opções para blob_get()

Em algumas situações, talvez seja necessário controlar exatamente o que blob_get tenta fazer usando os decoderparâmetros , compression e options .

O decodificador pode ser definido como auto (padrão) ou qualquer um dos seguintes valores:

compression pode ser ( auto padrão) none ou gzip.

Finalmente, o options parâmetro é do tipo jsonb. Existem quatro funções de utilidade que ajudam a construir valores para ele. Cada função do utilitário é designada para o descodificador correspondente ao seu nome.

Ao olhar para as definições de função, você pode ver quais parâmetros são suportados por qual decodificador.

options_csv_get - delimitador, null_string, cabeçalho, citação, escape, force_not_null, force_null, content_encoding options_tsv - delimitador, null_string, content_encoding options_copy - delimitador, null_string, cabeçalho, citação, escape, force_quote, force_not_null, force_null, content_encoding. options_binary - content_encoding

Sabendo o acima, podemos descartar gravações com null gravatar_id durante a análise.

INSERT INTO github_users
     SELECT user_id, url, UPPER(login), avatar_url, gravatar_id, display_login
       FROM azure_storage.blob_get('pgquickstart', 'github', 'users.csv.gz', NULL::github_users,
                                    options := azure_storage.options_csv_get(force_not_null := ARRAY['gravatar_id']));

INSERT 0 264308

Acesso ao armazenamento privado

1. Obtenha o nome da sua conta e a chave de acesso

   Sem uma chave de acesso, não teremos permissão para listar contêineres definidos como níveis de acesso Privado ou Blob.

   SELECT * FROM azure_storage.blob_list('mystorageaccount','privdatasets');
   

   ERROR:  azure_storage: missing account access key
   HINT:  Use SELECT azure_storage.account_add('<account name>', '<access key>')
   

   Na sua conta de armazenamento, abra as chaves de acesso. Copie o nome da conta de armazenamento e copie a seção Chave da chave1 (você deve selecionar Mostrar ao lado da chave primeiro).

   

2. Adicionar uma conta ao pg_azure_storage

   SELECT azure_storage.account_add('mystorageaccount', 'SECRET_ACCESS_KEY');
   

   Agora você pode listar contêineres definidos como níveis de acesso Privado e Blob para esse armazenamento, mas apenas como o citus usuário, que tem a azure_storage_admin função concedida a ele. Se você criar um novo usuário chamado support, ele não terá permissão para acessar o conteúdo do contêiner por padrão.

   SELECT * FROM azure_storage.blob_list('pgabs','dataverse');
   

   ERROR:  azure_storage: current user support is not allowed to use storage account pgabs
   

3. Permitir que o support usuário use uma conta específica do Armazenamento de Blob do Azure

   Conceder a permissão é tão simples quanto ligar para account_user_add.

   SELECT * FROM azure_storage.account_user_add('mystorageaccount', 'support');
   

   Podemos ver os usuários permitidos na saída do , que mostra todas as contas com chaves de account_listacesso definidas.

   SELECT * FROM azure_storage.account_list();
   

    account_name     | allowed_users
   ------------------+---------------
    mystorageaccount | {support}
   (1 row)
   

   Se você decidir, que o usuário não deve mais ter acesso. Basta ligar account_user_removepara .

   SELECT * FROM azure_storage.account_user_remove('mystorageaccount', 'support');
   

Próximos passos

Parabéns, você acabou de aprender como carregar dados no Azure Cosmos DB para PostgreSQL diretamente do Armazenamento de Blobs do Azure.

• Saiba como criar um painel em tempo real com o Azure Cosmos DB para PostgreSQL.
• Saiba mais sobre pg_azure_storage.
• Saiba mais sobre o suporte Postgres COPY.