Kurz: Vytvoření více toků s různými parametry

Kanál může obsahovat více toků, které jsou téměř identické a liší se pouze několika parametry. Definování těchto toků je explicitně náchylné k chybám, redundantní a obtížně se udržuje. Metaprogramování pomocí Python vnitřních funkcí generuje opakující se toky dynamicky, přičemž každé vyvolání poskytuje jinou sadu parametrů.

Přehled metaprogramování

Metaprogramování v deklarativních potrubí Lakeflow Spark používá vnitřní funkce Pythonu. Vzhledem k tomu, že tyto funkce jsou vyhodnocovány s odkladem při běhu kanálu, můžete zabalit @dp.table dekorátory uvnitř tovární funkce a volat tuto továrnu vícekrát s různými argumenty. Každé volání zaregistruje nový tok bez duplikování kódu.

Více informací o používání for smyček s deklarativními kanály Lakeflow Spark najdete v části Vytváření tabulek ve smyčcefor.

Příklad: Doby odezvy hasičského sboru

Následující příklad používá integrovanou datovou sadu hasičského sboru k vyhledání sousedství s nejrychlejší dobou odezvy tísňového volání pro každý typ volání. Bez metaprogramování musíte napsat téměř identické definice tabulky pro každý typ volání (Alarmy, Požár struktury, Lékařský incident). Při metaprogramování vygeneruje jedna tovární funkce všechny.

Krok 1: Definování nezpracované tabulky příjmu dat

import functools
from pyspark import pipelines as dp
from pyspark.sql.functions import *

@dp.table(
  name="raw_fire_department",
  comment="raw table for fire department response"
)
@dp.expect_or_drop("valid_received", "received IS NOT NULL")
@dp.expect_or_drop("valid_response", "responded IS NOT NULL")
@dp.expect_or_drop("valid_neighborhood", "neighborhood != 'None'")
def get_raw_fire_department():
  return (
    spark.read.format('csv')
      .option('header', 'true')
      .option('multiline', 'true')
      .load('/databricks-datasets/timeseries/Fires/Fire_Department_Calls_for_Service.csv')
      .withColumnRenamed('Call Type', 'call_type')
      .withColumnRenamed('Received DtTm', 'received')
      .withColumnRenamed('Response DtTm', 'responded')
      .withColumnRenamed('Neighborhooods - Analysis Boundaries', 'neighborhood')
      .select('call_type', 'received', 'responded', 'neighborhood')
  )

Krok 2: Definování funkce toku továrny

Funkce generate_tables továrny registruje dvě tabulky pro každý typ volání: filtrovanou tabulku volání a seřazenou tabulku doby odezvy. Obě jsou vytvořeny jako vnitřní funkce dekorované s @dp.table.

all_tables = []

def generate_tables(call_table, response_table, filter):
  @dp.table(
    name=call_table,
    comment="top level tables by call type"
  )
  def create_call_table():
    return spark.sql("""
      SELECT
        unix_timestamp(received,'M/d/yyyy h:m:s a') as ts_received,
        unix_timestamp(responded,'M/d/yyyy h:m:s a') as ts_responded,
        neighborhood
      FROM raw_fire_department
      WHERE call_type = '{filter}'
    """.format(filter=filter))

  @dp.table(
    name=response_table,
    comment="top 10 neighborhoods with fastest response time"
  )
  def create_response_table():
    return spark.sql("""
      SELECT
        neighborhood,
        AVG((ts_received - ts_responded)) as response_time
      FROM {call_table}
      GROUP BY 1
      ORDER BY response_time
      LIMIT 10
    """.format(call_table=call_table))

  all_tables.append(response_table)

Krok 3: Vyvolání továrny a definování souhrnné tabulky

Zavolejte továrnu jednou pro každý typ volání a pak definujte souhrnnou tabulku, která sjednocuje výsledky a vyhledá okolí, která se nejčastěji zobrazují napříč všemi kategoriemi.

generate_tables("alarms_table", "alarms_response", "Alarms")
generate_tables("fire_table", "fire_response", "Structure Fire")
generate_tables("medical_table", "medical_response", "Medical Incident")

@dp.table(
  name="best_neighborhoods",
  comment="which neighbor appears in the best response time list the most"
)
def summary():
  target_tables = [dp.read(t) for t in all_tables]
  unioned = functools.reduce(lambda x, y: x.union(y), target_tables)
  return (
    unioned.groupBy(col("neighborhood"))
      .agg(count("*").alias("score"))
      .orderBy(desc("score"))
  )

Poté, co spustíte tento pipeline, vytvoříte sadu podobných tabulek, jako v tomto grafu:

Graf tabulek vygenerovaných tímto kurzem

Klíčové koncepty

  • Vnitřní funkce jsou opožděně zaregistrovány: @dp.table Dekorátor nespustí funkci okamžitě. Zaregistruje funkci v pipeline runtime, který před zahájením spuštění provede kompletní zpracování grafu toku dat.
  • Uzavření zachycuje parametry: Každá vnitřní funkce se zavře nad parametry předanými do továrny (call_table, response_table, filter), takže každý registrovaný tok používá svou vlastní izolovanou sadu hodnot.
  • Dynamické seznamy tabulek: Použitím seznamu, jako je all_tables, ke sledování názvů tabulek generovaných programově je pozdější odkazování na ně snadné (například ve sjednocení nebo spojení).

Dodatečné zdroje

  • Last updated on