데코레이터는 @dp.foreach_sink() 싱크를 정의 ForEach 하는데, 이는 커스텀 Python 로직으로 한 행씩 스트림을 처리합니다. 데이터를 쓰기 위해 싱크를 추가 흐름이나 업데이트 흐름으로 target 참조합니다. 개념적 안내, 고려사항 및 예시는 '파이프라인의 각 행을 싱크로 쓰 ForEach 기'를 참조하세요. 각 마이크로 배치를 데이터프레임으로 처리하려면 foreach_batch_sink를 참조하세요.
Syntax
함수에서 싱크를 ForEach 정의하려면, 함수 @dp.foreach_sink()를 다음과 같이 장식합니다:
from pyspark import pipelines as dp
@dp.foreach_sink(name="<name>")
def row_handler(row):
"""
Required:
- `row`: a Row that contains one record of the stream.
"""
# Your custom write logic here.
# This function runs on the executors, so import any libraries it uses inside the function.
라이터 객체에서 싱크를 ForEach 정의하려면, 다음 writer 인자를 사용하여 객체를 호출 dp.foreach_sink() 하고 전달합니다:
from pyspark import pipelines as dp
class RowWriter:
def open(self, partition_id, epoch_id):
# Optional. Return False to skip the rows of this partition.
return True
def process(self, row):
# Required. Called for each row.
...
def close(self, error):
# Optional. `error` is None if processing succeeded.
...
dp.foreach_sink(name="<name>", writer=RowWriter())
예를 들어, decorator 대신 인수와 func 함께 함수 dp.foreach_sink() 를 전달할 수도 있습니다.dp.foreach_sink(name="<name>", func=row_handler)
Parameters
| Parameter | Description |
|---|---|
| 행 | Required.
Row A 는 스트림의 한 레코드를 포함합니다. 당신의 기능, 즉 process 작성자의 메서드는 각 통화마다 한 행씩 받습니다. |
| name | Optional. 파이프라인 내에서 싱크를 식별하는 고유한 이름입니다. 함수 이름으로 기본값이 됩니다. 라이터 객체의 경우, 기본적으로 소문자 클래스 이름을 사용합니다. |
| 함수 | Optional. 파이프라인이 각 행에 대해 호출하는 함수입니다. 직접 전화 dp.foreach_sink() 할 때 인 func 테리어 디자이너로 사용하지 마세요. 두 가지 모두 funcwriter통과할 수 없습니다. |
| writer | Optional. 메서드를 구현하는 객체이며, 선택적으로 and close(error) 메서드를 open(partition_id, epoch_id) 구현 process(row) 합니다. 키워드 논증으로 전달하세요. 두 가지 모두 funcwriter통과할 수 없습니다. |
작가 방법
라이터 객체는 Apache Spark Structured Streaming foreach 라이터와 동일한 메서드를 사용합니다. 각 마이크로 배치의 각 파티션에 대해 파이프라인은 를 호출open한 후, 만약 를 반환하면 openTrue각 행에 대해 호출 process 한 후, 다시 를 호출한다close. 구조화된 스트리밍에서 맞춤 출력 작성에 관한 정보는 임 의의 데이터 싱크에 쓰기 위한 foreachBatch 사용을 참조하세요.
| Method | Description |
|---|---|
open(partition_id, epoch_id) |
Optional. 파이프라인은 마이크로 배치의 파티션 행을 처리하기 전에 이 메서드를 호출합니다. 연결을 열거나 거래를 시작하는 데 사용하세요. 파티션 False 의 행을 처리하거나 건너뛰기 위해 돌아갑니다True. 작성자가 를 정의 open하지 않으면 파이프라인은 각 파티션의 행을 처리합니다.
partition_id
epoch_id 그리고 동일한 싱크에 쓰는 플로우 간에 반복될 수 있으며, 완전한 갱신 후에도 가능합니다. 행을 중복 제거하기 전에 ' 실패 처리와 중복 행'을 참고하세요. |
process(row) |
Required. 파이프라인은 각 행마다 이 메서드를 호출합니다. |
close(error) |
Optional. 파이프라인은 파티션의 행을 처리한 후 이 메서드를 호출합니다.
error 는 행을 처리하는 동안 발생한 예외 또는 None 처리가 성공했을 때 발생합니다. 연결을 닫는 데 사용하세요. 트랜잭션을 시작했다면open, 가 가 될 None때 error 만 커밋하고, 그렇지 않으면 롤백하세요. |