replace_flow

Important

이 기능은 베타 버전으로 제공됩니다.

데코레이터는 @dp.replace_flow 파이프라인 내 스트리밍 테이블에 대해 REPLACE USING flow를 생성합니다. 각 업데이트마다 플로우는 타겟 테이블에서 키 열과 replace_using 일치하는 모든 행을 교체하고, 나머지 행은 그대로 둡니다. 함수는 Apache Spark 스트리밍 데이터 프레임을 반환해야 합니다. 부분 스냅샷 대체를 REPLACE USING 플로우로 참고하세요.

소스가 열별로 키잉된 부분 스냅샷 연속일 때 사용 @dp.replace_flow 하세요. 대상 테이블과 플로우를 단일 문장으로 정의하려면 와 sequence_by 를 @dp.table에 전달 replace_using 하세요.

구문

from pyspark import pipelines as dp

dp.create_streaming_table("<target-table-name>") # Required only if the target table doesn't exist.

@dp.replace_flow(
  target = "<target-table-name>",
  replace_using = ["<key-column>", "<key-column>"],
  sequence_by = "<sequence-column>",
  name = "<flow-name>", # optional, defaults to function name
  comment = "<comment>", # optional
  spark_conf = {"<key>" : "<value>", "<key>" : "<value>"}) # optional
def <function-name>():
  return (<streaming-query>)

Parameters

매개 변수 Type Description
기능 function 필수 사항입니다. 사용자 정의 쿼리에서 Apache Spark 스트리밍 DataFrame을 반환하는 함수입니다.
target str 필수 사항입니다. 플로우의 타겟인 스트리밍 테이블의 이름입니다.
replace_using list 필수 사항입니다. 어떤 대상 행을 교체할지 식별하는 키 열들입니다. 최소한 한 개의 열을 지정하세요. 키 열은 반복할 수 없으며, 각 키 열의 타입은 정렬 가능해야 합니다.
sequence_by str 또는 Column 필수 사항입니다. 업데이트를 명령하는 열입니다. 각 키마다 가장 높은 시퀀스가 승리하며, 더 낮은 시퀀스 행은 이미 목표 내에 있는 더 높은 시퀀스를 덮어쓰지 않습니다.
name str 흐름 이름입니다. 제공되지 않으면 기본적으로 함수 이름이 지정됩니다.
comment str 흐름에 대한 설명입니다.
spark_conf dict 이 쿼리를 실행하기 위한 Spark 구성 목록입니다.

예제

from pyspark import pipelines as dp

# Keep the latest row for each order from a stream of partial snapshots
dp.create_streaming_table("orders_current")

@dp.replace_flow(
  target = "orders_current",
  replace_using = ["order_id"],
  sequence_by = "updated_at"
)
def orders_flow():
  return spark.readStream.table("order_updates")

레코드가 여러 열의 조합으로 식별될 때는 두 개 이상의 키 열을 사용하세요:

from pyspark import pipelines as dp

dp.create_streaming_table("accounts_current")

@dp.replace_flow(
  target = "accounts_current",
  replace_using = ["region", "account_id"],
  sequence_by = "updated_at"
)
def accounts_flow():
  return spark.readStream.table("account_updates")

Limitations

  • 스트리밍 테이블은 단일 REPLACE USING 플로우를 지원하며, 추가 플로우, 자동 CDC 플로 REPLACE WHERE 우, 플로우 등 다른 플로우 유형과 결합 REPLACE USING 할 수 없습니다.
  • 쿼리는 스트리밍 쿼리여야 합니다. @dp.replace_flow 비스트리밍 소스를 거부합니다.
  • REPLACE USING 플로우는 Databricks Runtime 18.2 이상이 필요합니다.
  • Last updated on