Cómo ingerir datos usando pg_azure_storage en Azure Cosmos DB for PostgreSQL

SE APLICA A: Azure Cosmos DB for PostgreSQL (con tecnología de la extensión de base de datos de Citus en PostgreSQL)

En este artículo se muestra cómo usar la extensión pg_azure_storage de PostgreSQL para manipular y cargar datos en Azure Cosmos DB for PostgreSQL directamente desde Azure Blob Storage (ABS). ABS es un servicio de almacenamiento nativo de nube escalable, duradero y seguro. Estas características lo convierten en una buena opción para almacenar y mover datos existentes a la nube.

Preparación de la base de datos y el almacenamiento de blobs

Para cargar datos desde Azure Blob Storage, instale la extensión pg_azure_storage PostgreSQL en su base de datos:

SELECT * FROM create_extension('azure_storage');

Importante

La extensión pg_azure_storage solo está disponible en clústeres de Azure Cosmos DB for PostgreSQL que ejecutan PostgreSQL 13 y versiones posteriores.

Hemos preparado un conjunto de datos de demostración pública para este artículo. Para usar su propio conjunto de datos, siga migración de sus datos locales al almacenamiento en la nube para aprender a introducir sus conjuntos de datos de forma eficaz en Azure Blob Storage.

Nota:

Al seleccionar "Contenedor (acceso de lectura anónimo para contenedores y blobs)", podrá ingerir archivos de Azure Blob Storage usando sus direcciones URL públicas y enumerar el contenido del contenedor sin necesidad de configurar una clave de cuenta en pg_azure_storage. Los contenedores fijados en el nivel de acceso "Privado (sin acceso anónimo)" o "Blob (acceso de lectura anónimo solo para blobs)" requerirán una clave de acceso.

Enumerar el contenido del contenedor

Existe una demostración de la cuenta de Azure Blob Storage y el contenedor creados previamente para este procedimiento. El nombre del contenedor es githuby está en la pgquickstart cuenta. Podemos ver fácilmente qué archivos están presentes en el contenedor mediante la azure_storage.blob_list(account, container) función.

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

Puede filtrar la salida mediante una WHERE cláusula SQL normal o mediante el prefix parámetro de la blob_list UDF. Esta última filtra las filas devueltas en Azure Blob Storage.

Nota:

Enumerar el contenido del contenedor requiere una cuenta y una clave de acceso o un contenedor con acceso anónimo habilitado.

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

Carga de datos desde el ABS

Carga de datos con el comando COPY

Empiece por crear un esquema de ejemplo.

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');

Cargar datos en las tablas es tan sencillo como llamar al 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 cómo la extensión ha reconocido que las direcciones URL proporcionadas al comando de copia son de Azure Blob Storage, que los archivos objetivo se han comprimido en gzip, y que además eso se controló automáticamente.

El COPY comando admite más parámetros y formatos. En el ejemplo anterior, el formato y la compresión se seleccionaron automáticamente en función de las extensiones de archivo. Sin embargo, puede proporcionar el formato directamente similar al COPY comando normal.

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

Actualmente, la extensión admite los siguientes formatos de archivo:

format	description
csv	Formato de valores separados por comas usado por COPY de PostgreSQL
tsv	Valores separados por tabulaciones, el formato COPY predeterminado de PostgreSQL
binary	Formato COPY de PostgreSQL binario
text	Un archivo que contiene un único valor de texto (por ejemplo, JSON grande o XML)

Carga de datos con blob_get()

El COPY comando es cómodo, pero tiene flexibilidad limitada. COPY usa internamente la función blob_get, que puede usar directamente para manipular datos en escenarios más complejos.

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

Nota

En la consulta anterior, el archivo se captura completamente antes de aplicarse LIMIT 3.

Con esta función, puede manipular datos sobre la marcha en consultas complejas y realizar importaciones 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

En el comando anterior, filtramos los datos a las cuentas con un gravatar_id presente y escribimos en mayúsculas sus inicios de sesión sobre la marcha.

Opciones de blob_get()

En algunas situaciones, es posible que tenga que controlar exactamente lo que blob_get intenta hacer mediante los decoder, compression y optionsparámetros.

El decodificador se puede establecer en auto (valor predeterminado) o en cualquiera de los siguientes valores:

format	description
csv	Formato de valores separados por comas usado por COPY de PostgreSQL
tsv	Valores separados por tabulaciones, el formato COPY predeterminado de PostgreSQL
binary	Formato COPY de PostgreSQL binario
text	Un archivo que contiene un único valor de texto (por ejemplo, JSON grande o XML)

compression puede ser auto (valor predeterminado), none o gzip.

Por último, el options parámetro es de tipo jsonb. Hay cuatro funciones de utilidad que ayudan a crear valores. Cada función de utilidad se designa para el decodificador que coincide con su nombre.

decodificador	función de utilidad
csv	options_csv_get
tsv	options_tsv
binary	options_binary
text	options_copy

Al examinar las definiciones de función, puede ver qué parámetros son compatibles con cada decodificador.

options_csv_get - delimitador, null_string, encabezado, comillas, escape, force_not_null, force_null, content_encoding options_tsv - delimitador, null_string, content_encoding options_copy - delimitador, null_string, encabezado, comillas, escape, force_quote, force_not_null, force_null, content_encoding. options_binary - content_encoding

Sabiendo lo anterior, podemos descartar las grabaciones con null gravatar_id durante el análisis.

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

Acceso al almacenamiento privado

1. Obtención del nombre de la cuenta y la clave de acceso

   Sin una clave de acceso, no se permitirá enumerar los contenedores que están establecidos en niveles de acceso Privado o de 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>')
   

   En la cuenta de almacenamiento, abra Claves de acceso. Copie el Nombre de la cuenta de almacenamiento y copie la sección Clave de key1 (primero debe seleccionar Mostrar junto a la clave).

   

2. Agregar una cuenta a pg_azure_storage

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

   Ahora puede enumerar los contenedores establecidos en niveles de acceso Blobs y Privados para ese almacenamiento, pero solo como el citus usuario, que tiene el azure_storage_admin rol concedido. Si crea un nuevo usuario denominado support, no podrá acceder al contenido del contenedor de forma predeterminada.

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

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

3. Permita que el support usuario use una cuenta específica de Azure Blob Storage

   Conceder el permiso es tan sencillo como llamar a account_user_add.

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

   Podemos ver los usuarios permitidos en la salida de account_list, que muestra todas las cuentas con claves de acceso definidas.

   SELECT * FROM azure_storage.account_list();
   

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

   Si alguna vez decide que el usuario ya no debe tener acceso. Simplemente llame a account_user_remove.

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

Pasos siguientes

Enhorabuena, acaba de aprender a cargar datos en Azure Cosmos DB for PostgreSQL directamente desde Azure Blob Storage.

• Obtenga información sobre cómo crear un panel en tiempo real con Azure Cosmos DB for PostgreSQL.
• Obtenga más información sobre pg_azure_storage.
• Obtenga más información sobre la compatibilidad con Postgres COPY.