Use uma tabela de pesquisa para grandes arrays de parâmetros em uma tarefa For each

For each As tarefas passam arrays de parâmetros para tarefas aninhadas de forma iterativa, para fornecer a cada tarefa aninhada informações para sua execução. Arrays de parâmetros são limitados a 5.000 caracteres, ou 48 KB se você usar referências de valor de tarefa para passá-los. Quando você tem uma quantidade maior de dados para passar para as tarefas aninhadas, não é possível usar diretamente a entrada, os valores da tarefa ou os parâmetros do job para passar esses dados.

Uma alternativa para passar os dados completos é armazenar os dados da tarefa como um arquivo JSON e passar uma key de pesquisa (para os dados JSON) através da entrada da tarefa, em vez dos dados completos. Tarefas aninhadas podem usar a key para recuperar os dados específicos necessários para cada iteração.

O exemplo a seguir mostra um arquivo de configuração JSON de exemplo e como passar parâmetros para uma tarefa aninhada que procura os valores na configuração JSON.

Configuração JSON de amostra

Esta configuração de exemplo é uma lista de passos, com parâmetros (args) para cada iteração (apenas três passos são mostrados para este exemplo). Suponha que este arquivo JSON seja salvo como /Workspace/Users/<user>/copy-filtered-table-config.json. Fazemos referência a isso dentro da tarefa aninhada.

JSON

{
  "steps": [
    {
      "key": "table_1",
      "args": {
        "catalog": "my-catalog",
        "schema": "my-schema",
        "source_table": "raw_data_table_1",
        "destination_table": "filtered_table_1",
        "filter_column": "col_a",
        "filter_value": "value_1"
      }
    },
    {
      "key": "table_2",
      "args": {
        "catalog": "my-catalog",
        "schema": "my-schema",
        "source_table": "raw_data_table_2",
        "destination_table": "filtered_table_2",
        "filter_column": "col_b",
        "filter_value": "value_2"
      }
    },
    {
      "key": "table_3",
      "args": {
        "catalog": "my-catalog",
        "schema": "my-schema",
        "source_table": "raw_data_table_3",
        "destination_table": "filtered_table_3",
        "filter_column": "col_c",
        "filter_value": "value_3"
      }
    }
  ]
}

Tarefa de exemplo For each

A tarefa For each em seu Job inclui entrada com as key para cada iteração. Este exemplo mostra uma tarefa chamada copy-filtered-tables com as Entradas definidas como ["table_1","table_2","table_3"]. Esta lista é limitada a 5.000 caracteres, mas como você está apenas passando chaves, ela é muito menor do que os dados completos.

Neste exemplo, os passos não dependem de outros passos ou tarefas, então podemos definir uma concorrência maior que 1 para fazer a tarefa ser executada mais rapidamente.

Tarefa aninhada de exemplo

A tarefa aninhada recebe a entrada da tarefa pai For each. Neste caso, configuramos a entrada para ser usada como um Key para o arquivo de configuração. A imagem a seguir mostra a tarefa aninhada, incluindo a configuração de um Parâmetro chamado key com o valor {{input}}.

Esta tarefa é um Notebook que contém código. Em seu Notebook, você pode usar o seguinte código Python para ler a entrada e usá-la como uma key no arquivo JSON de configuração. Os dados do arquivo JSON são usados para ler, filtrar e gravar dados de uma tabela.

nota

Para ver um exemplo de como acionar uma For each tarefa a partir de uma tabela Delta dinâmica em vez de um arquivo JSON estático, consulte Usar uma tabela de controle para acionar um For each Job.

Python

# copy-filtered-table (iteratable task code to read a table, filter by a value, and write as a new table)

from pyspark.sql.functions import expr
from types import SimpleNamespace

import json

# If the notebook is run outside of a job with a key parameter, this provides
# a default. This allows testing outside of a For each task
dbutils.widgets.text("key", "table_1", "key")

# load configuration (note that the path must be set to valid configuration file)
config_path = "/Workspace/Users/<user>/copy-filtered-table-config.json"
with open(config_path, "r") as file:
    config = json.loads(file.read())

# look up step and arguments
key = dbutils.widgets.get("key")
current_step = next((step for step in config['steps'] if step['key'] == key), None)
if current_step is None:
    raise ValueError(f"Could not find step '{key}' in the configuration")

args = SimpleNamespace(**current_step["args"])

# read the source table defined for the step, and filter it
df = spark.read.table(f"{args.catalog}.{args.schema}.{args.source_table}") \
          .filter(expr(f"{args.filter_column} like '%{args.filter_value}%'"))

# write the filtered table to the destination
df.write.mode("overwrite").saveAsTable(f"{args.catalog}.{args.schema}.{args.destination_table}")