foreach_sink

Dekoratör, @dp.foreach_sink() özel Python mantığı ile bir akışı satır bir satırda işleyen bir ForEach sink tanımlar. Sink'i target bir ek akışta veya veri yazmak için güncelleme akışında referans verirsiniz. Kavramsal rehberlik, dikkate alınanlar ve örnekler için, bkz. Her satırı pipelines'ta yazmak için bir ForEach sink kullanın. Her mikro-partiyi DataFrame olarak işlemek için bkz. foreach_batch_sink.

Syntax

Bir fonksiyondan bir ForEach lavabo tanımlamak için, fonksiyonu şu şekilde @dp.foreach_sink()süsleyin:

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.

Bir yazıcı nesnesinden bir ForEach sink tanımlamak için, nesneyi argümanla writer birlikte çağırın dp.foreach_sink() ve geçirin:

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

Ayrıca argümanla bir fonksiyonu dp.foreach_sink()func dekoratör yerine aktarabilirsiniz, örneğin dp.foreach_sink(name="<name>", func=row_handler).

Parameters

Parametre	Description
satır	Required. Yayının bir kaydını içeren A Row . Sizin işleviniz veya process yazarınızın yöntemi, her arama için bir satır alır.
name	Optional. İşlem hattı içindeki havuzu tanımlamak için benzersiz bir ad. Varsayılan olarak fonksiyonun adı kalır. Bir yazar nesnesi için, sınıf adı küçük harflerle varsayılan olarak kullanılır.
func	Optional. Boru hattının her satır için çağırdığı fonksiyon. func Doğrudan aradığınızda dp.foreach_sink() kullanın, dekoratör olarak değil. İkisini func de geçemezsiniz ve writer.
writer	Optional. Metodu process(row) uygulayan bir nesne ve isteğe bağlı open(partition_id, epoch_id) olarak ve close(error) metodları. Bunu anahtar kelime argümanı olarak geçir. İkisini func de geçemezsiniz ve writer.

Yazar yöntemleri

Bir yazar nesnesi, Apache Spark Yapılandırılmış Akış foreach yazarıyla aynı yöntemleri kullanır. Her mikro-batch'in her bölümü için, pipeline open, process sonra döndürürse openTrueher satır için çağırır ve closeardından . Structured Streaming'de özel çıktı yazma hakkında bilgi için Use foreachBatch to write to random data sins'e bakınız.

Method	Description
open(partition_id, epoch_id)	Optional. Boru hattı, mikro parti için bir bölümün satırlarını işlemeden önce bu yöntemi çağırır. Bağlantı açmak veya işlemler başlatmak için kullanın. Bölüm satırlarını işlemek için geri True dön ya da False atla. Yazar tanımlamazsa open, pipeline her bölümün satırlarını işliyor. partition_id ve epoch_id aynı yıkamaya ve tam yenilemeye giden akışlar arasında tekrarlanabilir. Satır deduplicate için kullanmadan önce, Hataları ve Kopya Satırları Yönet bölümünü inceleyin.
process(row)	Required. Boru hattı her satır için bu yöntemi çağırır.
close(error)	Optional. Boru hattı, bölüm satırlarını işledikten sonra bu yöntemi çağırır. error satır işlenirken veya None işlemin başarılı olması durumunda meydana gelen istisnadır. Bağlantıları kapatmak için kullanın. Bir işlem başlattıysa open , sadece error olduğunda Nonecommit yapın ve aksi takdirde geri alın.