foreach_sink

Декоратор @dp.foreach_sink() определяет синкForEach, который обрабатывает поток по одной строке с пользовательской логикой Python. Вы ссылаетесь на поsink как target на дополнительный поток или как на update flow для записи данных. Для концептуальных рекомендаций, соображдений и примеров см. раздел «Использовать ForEach посылку для записи каждой строки в конвейерах». Чтобы обработать каждую микропартию как DataFrame, см. 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 sink из объекта writer, вызовите dp.foreach_sink() и передайте объект с аргументом writer :

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())

Вы также можете передать функцию с dp.foreach_sink()func аргументом вместо использования декоратора, например dp.foreach_sink(name="<name>", func=row_handler), .

Parameters

Parameter	Description
строка	Required. В Row котором содержится одна запись потока. Ваша функция, или process метод вашего автора, получает одну строку за каждый звонок.
name	Optional. Уникальное имя для идентификации приемника в конвейере. По умолчанию используется имя функции. Для объекта writer по умолчанию имя класса используется в строчной букве.
func	Optional. Функция, которую конвейер вызывает для каждой строки. Используйте func их при звонке dp.foreach_sink() напрямую, а не как декоратора. Нельзя пройти оба func и writer.
writer	Optional. Объект, реализующий метод process(row) , а также open(partition_id, epoch_id) опционально методы и методы close(error) . Передайте это как аргумент по ключевым словам. Нельзя пройти оба func и writer.

Методы писателя

Объект writer использует те же методы, что и Apache Spark Structured Streaming foreach writer. Для каждого раздела каждой микропартии конвейер вызывает open, затем вызывает process каждую строку, если open возвращает True, и затем вызывает close. Для получения информации о написании пользовательских выходов в структурированном потоковом процессе см. раздел «Использовать foreachBatch для записи в произвольные погрузчики данных».

Метод	Description
open(partition_id, epoch_id)	Optional. Конвейер вызывает этот метод до обработки строк раздела для микропакета. Используйте его для открытия соединений или запуска транзакций. Возвращайтесь True , чтобы обработать строки разбиения или False пропустить их. Если автор не определяет open, конвейер обрабатывает строки каждого раздела. partition_id и epoch_id может повторять процесс между потоками, которые записываются в один и тот же умывал, после полного обновления. Прежде чем использовать их для удаления строк, смотрите раздел «Обработка отказов» и «Дублирование строк».
process(row)	Required. Конвейер вызывает этот метод для каждой строки.
close(error)	Optional. Конвейер вызывает этот метод после обработки строк раздела. error является исключением, возникшим при обработке строк или None если обработка прошла успешно. Используйте его, чтобы закрыть соединения. Если open транзакция запущена, коммит только когда error , и Noneоткат назад, иначе.