Partilhar via

Tutorial: COPY INTO com o Spark SQL

O Databricks recomenda que você use o comando COPY INTO para carregamento de dados incrementais e em massa para fontes de dados que contenham milhares de arquivos. A Databricks recomenda que tu uses Auto Loader em casos de uso avançados.

Neste tutorial, você usa o comando COPY INTO para carregar dados do armazenamento de objetos na nuvem em uma tabela no espaço de trabalho do Azure Databricks.

Requerimentos

Passo 1. Configure seu ambiente e crie um gerador de dados

Este tutorial pressupõe familiaridade básica com o Azure Databricks e uma configuração de espaço de trabalho padrão. Se não conseguir executar o código fornecido, contacte o administrador da área de trabalho para se certificar de que tem acesso a recursos de computação e a uma localização na qual possa escrever dados.

Observe que o código fornecido usa um source parâmetro para especificar o local que você configurará como sua COPY INTO fonte de dados. Conforme está escrito, este código aponta para um local na raiz do DBFS. Se você tiver permissões de gravação em um local de armazenamento de objetos externo, substitua a parte dbfs:/ da cadeia de caracteres de origem pelo caminho para o armazenamento de objetos. Como este bloco de código também faz uma exclusão recursiva para redefinir esta demonstração, certifique-se de não apontar este bloco para os dados de produção e de manter o diretório /user/{username}/copy-into-demo aninhado para evitar substituir ou eliminar dados existentes.

  1. Crie um novo bloco de notas e anexe-o a um recurso de computação.

  2. Copie e execute o código a seguir para redefinir o local de armazenamento e o banco de dados usados neste tutorial:

    %python
# Set parameters for isolation in workspace and reset demo

username = spark.sql("SELECT regexp_replace(current_user(), '[^a-zA-Z0-9]', '_')").first()[0]
database = f"copyinto_{username}_db"
source = f"dbfs:/user/{username}/copy-into-demo"

spark.sql(f"SET c.username='{username}'")
spark.sql(f"SET c.database={database}")
spark.sql(f"SET c.source='{source}'")

spark.sql("DROP DATABASE IF EXISTS ${c.database} CASCADE")
spark.sql("CREATE DATABASE ${c.database}")
spark.sql("USE ${c.database}")

dbutils.fs.rm(source, True)

  3. Copie e execute o código a seguir para configurar algumas tabelas e funções que serão usadas para gerar dados aleatoriamente:

    -- Configure random data generator

CREATE TABLE user_ping_raw
(user_id STRING, ping INTEGER, time TIMESTAMP)
USING json
LOCATION ${c.source};

CREATE TABLE user_ids (user_id STRING);

INSERT INTO user_ids VALUES
("potato_luver"),
("beanbag_lyfe"),
("default_username"),
("the_king"),
("n00b"),
("frodo"),
("data_the_kid"),
("el_matador"),
("the_wiz");

CREATE FUNCTION get_ping()
    RETURNS INT
    RETURN int(rand() * 250);

CREATE FUNCTION is_active()
    RETURNS BOOLEAN
    RETURN CASE
        WHEN rand() > .25 THEN true
        ELSE false
        END;

Etapa 2: Gravar os dados de exemplo no armazenamento em nuvem

Gravar em formatos de dados diferentes do Delta Lake é raro no Azure Databricks. O código fornecido aqui grava em JSON, simulando um sistema externo que pode despejar resultados de outro sistema no armazenamento de objetos.

  1. Copie e execute o seguinte código para escrever um lote de dados JSON brutos:

    -- Write a new batch of data to the data source

INSERT INTO user_ping_raw
SELECT *,
  get_ping() ping,
  current_timestamp() time
FROM user_ids
WHERE is_active()=true;

Etapa 3: Usar COPY INTO para carregar dados JSON idempotentes

Você deve criar uma tabela Delta Lake de destino antes de poder usar COPY INTO. Você não precisa fornecer nada além de um nome de tabela em sua CREATE TABLE declaração.

  1. Copie e execute o código a seguir para criar sua tabela Delta de destino e carregar dados de sua origem:

    -- Create target table and load data

CREATE TABLE IF NOT EXISTS user_ping_target;

COPY INTO user_ping_target
FROM ${c.source}
FILEFORMAT = JSON
FORMAT_OPTIONS ("mergeSchema" = "true")
COPY_OPTIONS ("mergeSchema" = "true")

Como essa ação é idempotente, você pode executá-la várias vezes, mas os dados só serão carregados uma vez.

Passo 4: Pré-visualizar o conteúdo da tabela

Você pode executar uma consulta SQL simples para revisar manualmente o conteúdo desta tabela.

  1. Copie e execute o seguinte código para visualizar sua tabela:

    -- Review updated table

SELECT * FROM user_ping_target

Etapa 5: carregar mais dados e visualizar resultados

Pode executar novamente as etapas 2 a 4 várias vezes para receber novos lotes de dados JSON brutos aleatórios na sua fonte, carregá-los de forma idempotente no Delta Lake com COPY INTOe pré-visualizar os resultados. Tente executar essas etapas fora de ordem ou várias vezes para simular vários lotes de dados brutos sendo gravados ou executados COPY INTO várias vezes sem que novos dados tenham chegado.

Passo 6: Tutorial de limpeza

Quando terminar este tutorial, você poderá limpar os recursos associados se não quiser mais mantê-los.

Copie e execute o seguinte código para soltar o banco de dados, tabelas e remover todos os dados:

%python
# Drop database and tables and remove data

spark.sql("DROP DATABASE IF EXISTS ${c.database} CASCADE")
dbutils.fs.rm(source, True)

Recursos adicionais

  • Last updated on