Partilhar via

Tutorial: COPY INTO com o Spark SQL

  • Artigo

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

  1. Uma assinatura do Azure, um espaço de trabalho do Azure Databricks nessa assinatura e um cluster nesse espaço de trabalho. Para criá-los, consulte Guia de início rápido: executar um trabalho do Spark no espaço de trabalho do Azure Databricks usando o portal do Azure. Se você seguir este início rápido, não precisará seguir as instruções na seção Executar um trabalho do Spark SQL.
  2. Um cluster versátil no seu espaço de trabalho a executar o Databricks Runtime 11.3 LTS ou superior. Para criar um cluster multiuso, consulte Guia de Configuração de Computação.
  3. Familiaridade com a interface do usuário do espaço de trabalho do Azure Databricks. Consulte para Navegar no espaço de trabalho.
  4. Familiaridade em trabalhar com notebooks Databricks.
  5. Um local no qual você pode gravar dados; esta demonstração usa a raiz DBFS como exemplo, mas o Databricks recomenda um local de armazenamento externo configurado com o Unity Catalog.

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 parâmetro source para especificar o local que você configurará como sua fonte de dados COPY INTO. 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 eliminação recursiva para reiniciar este exemplo, certifique-se de não o direcionar para dados de produção e de manter o diretório aninhado /user/{username}/copy-into-demo para evitar sobrepor ou eliminar dados existentes.

  1. Crie um novo notebook SQL e anexe-o a um cluster que esteja a executar o Databricks Runtime 11.3 LTS ou acima.

  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. No Databricks Runtime 11.3 LTS e superior, você não precisa fornecer nada além de um nome de tabela em sua instrução CREATE TABLE. Para versões anteriores do Databricks Runtime, você deve fornecer um esquema ao criar uma tabela vazia.

  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.

  1. 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)

  2. Para interromper o recurso de computação, vá para o separador Clusters e encerre o cluster.

Recursos adicionais