Padrões comuns de carregamento de dados

Auto Loader simplifica uma série de tarefas comuns de ingestão de dados. Essa referência rápida fornece exemplos de vários padrões populares.

Filtrando diretórios ou arquivos usando padrões globais

Os padrões Glob podem ser usados para filtrar diretórios e arquivos quando fornecidos no caminho.

Padrão	Descrição
?	Corresponde a qualquer caractere único
*	Corresponde a zero ou mais caracteres
[abc]	Corresponde a um único caractere do conjunto de caracteres {a, b, c}.
[a-z]	Corresponde a um único caractere do intervalo de caracteres {a... z}.
[^a]	Corresponde a um único caractere que não é do conjunto de caracteres ou do intervalo {a}. Observe que o caractere ^ deve ocorrer imediatamente à direita do colchete de abertura.
{ab,cd}	Corresponde a uma cadeia de caracteres do conjunto de cadeias de caracteres {ab, cd}.
{ab,c{de, fh}}	Corresponde a uma cadeia de caracteres do conjunto de cadeias de caracteres {ab, cde, cfh}.

Use o path para fornecer padrões de prefixo, por exemplo:

df = spark.readStream.format("cloudFiles") \
  .option("cloudFiles.format", <format>) \
  .schema(schema) \
  .load("<base-path>/*/files")

val df = spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", <format>)
  .schema(schema)
  .load("<base-path>/*/files")

Importante

Você precisa usar a opção pathGlobFilter para fornecer explicitamente padrões de sufixo. O path fornece apenas um filtro de prefixo.

Por exemplo, se você quiser analisar somente arquivos png em um diretório que contém arquivos com sufixos diferentes, você pode fazer:

df = spark.readStream.format("cloudFiles") \
  .option("cloudFiles.format", "binaryFile") \
  .option("pathGlobfilter", "*.png") \
  .load(<base-path>)

val df = spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", "binaryFile")
  .option("pathGlobfilter", "*.png")
  .load(<base-path>)

Observação

O comportamento de globbing do default do Auto Loader é diferente do comportamento do default de outras fontes de arquivos do Spark. Adicione .option("cloudFiles.useStrictGlobber", "true") à sua leitura para usar o globbing que corresponde ao comportamento do default Spark em relação às fontes de arquivos. Consulte a tabela a seguir para obter mais informações sobre globbing:

Padrão	Caminho do arquivo	padrão globber	Rigoroso globber
/a/b	/a/b/c/file.txt	Sim	Sim
/a/b	/a/b_dir/c/file.txt	Não	Não
/a/b	/a/b.txt	Não	Não
/a/b/	/a/b.txt	Não	Não
/a/ */c/	/a/b/c/file.txt	Sim	Sim
/a/ */c/	/a/b/c/d/file.txt	Sim	Sim
/a/ */c/	/a/b/x/y/c/file.txt	Sim	Não
/a/ */c	/a/b/c_file.txt	Sim	Não
/a/ */c/	/a/b/c_file.txt	Sim	Não
/a/ */c/	/a/ */cookie/file.txt	Sim	Não
/a/b *	/a/b.txt	Sim	Sim
/a/b *	/a/b/file.txt	Sim	Sim
/a/ {0.txt ,1.txt}	/a/0.txt	Sim	Sim
/a/ */ {0.txt ,1.txt}	/a/0.txt	Não	Não
/a/b/ [cde-h] /i/	/a/b/c/i/file.txt	Sim	Sim

Permitir ETL fácil

Uma maneira fácil de colocar seus dados no Delta Lake sem perder nenhum dado é usar o seguinte padrão e ativar a inferência de esquema com o Auto Loader. Databricks recomenda a execução do código a seguir em um trabalho Databricks para que ele reinicie automaticamente a transmissão quando o esquema dos dados de origem for alterado. Em default, o esquema é inferido como tipos de cadeia de caracteres, todos os erros de análise (não deve haver nenhum se tudo permanecer como cadeia de caracteres) irão para _rescued_data, e todas as novas colunas falharão na transmissão e evoluirão o esquema.

spark.readStream.format("cloudFiles") \
  .option("cloudFiles.format", "json") \
  .option("cloudFiles.schemaLocation", "<path-to-schema-location>") \
  .load("<path-to-source-data>") \
  .writeStream \
  .option("mergeSchema", "true") \
  .option("checkpointLocation", "<path-to-checkpoint>") \
  .start("<path_to_target")

spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", "json")
  .option("cloudFiles.schemaLocation", "<path-to-schema-location>")
  .load("<path-to-source-data>")
  .writeStream
  .option("mergeSchema", "true")
  .option("checkpointLocation", "<path-to-checkpoint>")
  .start("<path_to_target")

Evite a perda de dados em dados bem estruturados

Quando o senhor conhece o esquema, mas deseja saber sempre que receber dados inesperados, a Databricks recomenda o uso do rescuedDataColumn.

spark.readStream.format("cloudFiles") \
  .schema(expected_schema) \
  .option("cloudFiles.format", "json") \
  # will collect all new fields as well as data type mismatches in _rescued_data
  .option("cloudFiles.schemaEvolutionMode", "rescue") \
  .load("<path-to-source-data>") \
  .writeStream \
  .option("checkpointLocation", "<path-to-checkpoint>") \
  .start("<path_to_target")

spark.readStream.format("cloudFiles")
  .schema(expected_schema)
  .option("cloudFiles.format", "json")
  // will collect all new fields as well as data type mismatches in _rescued_data
  .option("cloudFiles.schemaEvolutionMode", "rescue")
  .load("<path-to-source-data>")
  .writeStream
  .option("checkpointLocation", "<path-to-checkpoint>")
  .start("<path_to_target")

Se quiser que a transmissão interrompa o processamento se for introduzido um novo campo que não corresponda ao seu esquema, o senhor pode adicionar:

.option("cloudFiles.schemaEvolutionMode", "failOnNewColumns")

Permitir um pipeline de dados semiestruturado e flexível

Quando o senhor recebe dados de um fornecedor que introduz novas colunas nas informações que ele fornece, pode não saber exatamente quando ele faz isso ou pode não ter a largura de banda necessária para atualizar seu pipeline de dados. Agora, o senhor pode aproveitar a evolução do esquema para reiniciar a transmissão e permitir que o site Auto Loader atualize o esquema inferido automaticamente. Você também pode aproveitar schemaHints para alguns dos campos “sem esquema” que o fornecedor pode estar fornecendo.

spark.readStream.format("cloudFiles") \
  .option("cloudFiles.format", "json") \
  # will ensure that the headers column gets processed as a map
  .option("cloudFiles.schemaHints",
          "headers map<string,string>, statusCode SHORT") \
  .load("/api/requests") \
  .writeStream \
  .option("mergeSchema", "true") \
  .option("checkpointLocation", "<path-to-checkpoint>") \
  .start("<path_to_target")

spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", "json")
  // will ensure that the headers column gets processed as a map
  .option("cloudFiles.schemaHints",
          "headers map<string,string>, statusCode SHORT")
  .load("/api/requests")
  .writeStream
  .option("mergeSchema", "true")
  .option("checkpointLocation", "<path-to-checkpoint>")
  .start("<path_to_target")

Transformar dados JSON aninhados

Como o site Auto Loader infere as colunas JSON de nível superior como strings, o senhor pode ficar com objetos JSON aninhados que exigem transformações adicionais. O senhor pode usar as APIs de acesso a dados semiestruturados para transformar ainda mais o conteúdo JSON complexo.

spark.readStream.format("cloudFiles") \
  .option("cloudFiles.format", "json") \
  # The schema location directory keeps track of your data schema over time
  .option("cloudFiles.schemaLocation", "<path-to-checkpoint>") \
  .load("<source-data-with-nested-json>") \
  .selectExpr(
    "*",
    "tags:page.name",    # extracts {"tags":{"page":{"name":...}}}
    "tags:page.id::int", # extracts {"tags":{"page":{"id":...}}} and casts to int
    "tags:eventType"     # extracts {"tags":{"eventType":...}}
  )

spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", "json")
  // The schema location directory keeps track of your data schema over time
  .option("cloudFiles.schemaLocation", "<path-to-checkpoint>")
  .load("<source-data-with-nested-json>")
  .selectExpr(
    "*",
    "tags:page.name",     // extracts {"tags":{"page":{"name":...}}}
    "tags:page.id::int",  // extracts {"tags":{"page":{"id":...}}} and casts to int
    "tags:eventType"      // extracts {"tags":{"eventType":...}}
  )

Inferir dados JSON aninhados

Quando você tem dados aninhados, você pode usar a opção cloudFiles.inferColumnTypes para inferir a estrutura aninhada de seus dados e outros tipos de coluna.

spark.readStream.format("cloudFiles") \
  .option("cloudFiles.format", "json") \
  # The schema location directory keeps track of your data schema over time
  .option("cloudFiles.schemaLocation", "<path-to-checkpoint>") \
  .option("cloudFiles.inferColumnTypes", "true") \
  .load("<source-data-with-nested-json>")

spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", "json")
  // The schema location directory keeps track of your data schema over time
  .option("cloudFiles.schemaLocation", "<path-to-checkpoint>")
  .option("cloudFiles.inferColumnTypes", "true")
  .load("<source-data-with-nested-json>")

Carregar arquivos CSV sem cabeçalhos

df = spark.readStream.format("cloudFiles") \
  .option("cloudFiles.format", "csv") \
  .option("rescuedDataColumn", "_rescued_data") \ # makes sure that you don't lose data
  .schema(<schema>) \ # provide a schema here for the files
  .load(<path>)

val df = spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", "csv")
  .option("rescuedDataColumn", "_rescued_data") // makes sure that you don't lose data
  .schema(<schema>) // provide a schema here for the files
  .load(<path>)

Aplicar um esquema em arquivos CSV com cabeçalhos

df = spark.readStream.format("cloudFiles") \
  .option("cloudFiles.format", "csv") \
  .option("header", "true") \
  .option("rescuedDataColumn", "_rescued_data") \ # makes sure that you don't lose data
  .schema(<schema>) \ # provide a schema here for the files
  .load(<path>)

val df = spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", "csv")
  .option("header", "true")
  .option("rescuedDataColumn", "_rescued_data") // makes sure that you don't lose data
  .schema(<schema>) // provide a schema here for the files
  .load(<path>)

Ingerir dados binários ou de imagem no Delta Lake para ML

Depois que os dados são armazenados em Delta Lake, o senhor pode executar a inferência distribuída nos dados. Consulte Realizar inferência distribuída usando Pandas UDF .

spark.readStream.format("cloudFiles") \
  .option("cloudFiles.format", "binaryFile") \
  .load("<path-to-source-data>") \
  .writeStream \
  .option("checkpointLocation", "<path-to-checkpoint>") \
  .start("<path_to_target")

spark.readStream.format("cloudFiles")
  .option("cloudFiles.format", "binaryFile")
  .load("<path-to-source-data>")
  .writeStream
  .option("checkpointLocation", "<path-to-checkpoint>")
  .start("<path_to_target")

Sintaxe do Auto Loader para DLT

O Delta Live Tables fornece uma sintaxe Python ligeiramente modificada para o Auto Loader e adiciona suporte a SQL para o Auto Loader.

Os exemplos a seguir utilizam o Auto Loader para criar conjuntos de dados a partir de arquivos CSV e JSON:

@dlt.table
def customers():
  return (
    spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "csv")
      .load("/databricks-datasets/retail-org/customers/")
  )

@dlt.table
def sales_orders_raw():
  return (
    spark.readStream.format("cloudFiles")
      .option("cloudFiles.format", "json")
      .load("/databricks-datasets/retail-org/sales_orders/")
  )

CREATE OR REFRESH STREAMING TABLE customers
AS SELECT * FROM cloud_files("/databricks-datasets/retail-org/customers/", "csv")

CREATE OR REFRESH STREAMING TABLE sales_orders_raw
AS SELECT * FROM cloud_files("/databricks-datasets/retail-org/sales_orders/", "json")

O senhor pode usar as opções de formato compatíveis com o Auto Loader. Usando a função map(), você pode passar opções para o método cloud_files(). As opções são par key-value, em que a chave e os valores são strings. A seguir, descrevemos a sintaxe para trabalhar com o Auto Loader no SQL:

CREATE OR REFRESH STREAMING TABLE <table-name>
AS SELECT *
  FROM cloud_files(
    "<file-path>",
    "<file-format>",
    map(
      "<option-key>", "<option_value",
      "<option-key>", "<option_value",
      ...
    )
  )

O exemplo a seguir lê dados de arquivos CSV delimitados por tabulação com um cabeçalho:

CREATE OR REFRESH STREAMING TABLE customers
AS SELECT * FROM cloud_files("/databricks-datasets/retail-org/customers/", "csv", map("delimiter", "\t", "header", "true"))

Você pode utilizar o schema para especificar o formato manualmente; você deve especificar o schema para formatos que não suportam inferência de esquema:

@dlt.table
def wiki_raw():
  return (
    spark.readStream.format("cloudFiles")
      .schema("title STRING, id INT, revisionId INT, revisionTimestamp TIMESTAMP, revisionUsername STRING, revisionUsernameId INT, text STRING")
      .option("cloudFiles.format", "parquet")
      .load("/databricks-datasets/wikipedia-datasets/data-001/en_wikipedia/articles-only-parquet")
  )

CREATE OR REFRESH STREAMING TABLE wiki_raw
AS SELECT *
  FROM cloud_files(
    "/databricks-datasets/wikipedia-datasets/data-001/en_wikipedia/articles-only-parquet",
    "parquet",
    map("schema", "title STRING, id INT, revisionId INT, revisionTimestamp TIMESTAMP, revisionUsername STRING, revisionUsernameId INT, text STRING")
  )

Observação

O Delta Live Tables configura e gerencia automaticamente o esquema e os diretórios de ponto de verificação ao usar o Auto Loader para ler arquivos. No entanto, se você configurar manualmente um desses diretórios, executar uma atualização completa não afetará o conteúdo dos diretórios configurados. O Databricks recomenda a utilização dos diretórios configurados automaticamente como uma medida para prevenir efeitos secundários indesejados durante o processamento.