append_flow

Dekoratör, @dp.append_flow işlem hattı tablolarınız için ekleme akışları veya geri doldurmalar oluşturur. İşlev bir Apache Spark akış veri çerçevesi döndürmelidir. Bkz . Lakeflow işlem hattı akışlarıyla verileri artımlı olarak yükleme ve işleme.

Ekleyen akışlar, akış tablolarını, yönetilen tabloları veya sinkleri hedefleyebilir.

Sözdizimi

from pyspark import pipelines as dp

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

@dp.append_flow(
  target = "<target-table-name>",
  name = "<flow-name>", # optional, defaults to function name
  once = False, # optional
  depends_on = "<flow-name>", # optional, Public Preview
  spark_conf = {"<key>" : "<value", "<key" : "<value>"}, # optional
  comment = "<comment>", # optional
  import_checkpoint = "<checkpoint-path>") # optional
def <function-name>():
  return (<streaming-query>) #

Parametreler

Parametre Türü Description
function function Gerekli. Kullanıcı tanımlı bir sorgudan Apache Spark akış DataFrame'i döndüren işlev.
target str Gerekli. Ekleme akışının hedefi olan tablo veya havuzun adı.
name str Akış adı. Sağlanmadıysa, varsayılan olarak işlev adını kullanır.
once bool İsteğe bağlı olarak, akışı yedek doldurma gibi tek seferlik bir akış olarak tanımlayın. Kullanımı once=True , akışı iki şekilde değiştirir:
  • Geri dönüş değeri. streaming-query. Bu durumda, bir akış DataFrame değil, bir yığın DataFrame olmalıdır.
  • Akış varsayılan olarak bir kez çalıştırılır. Eğer işlem hattı eksiksiz bir yenilemeyle güncellenirse, ONCE akış verileri yeniden oluşturmak için tekrar çalıştırılır.
depends_on str veya list Genel Önizleme. Bu akışın başlamasından önce başarıyla tamamlanması gereken bir veya daha fazla akış adı. Tek bir akış adını veya isim listesini kabul eder. Sürekli bir boru hattında, her selef tek seferlik (once) akış olmalıdır. Bu emir akışı sadece yürütülür; akışın nasıl çalıştığını değiştirmez. Bkz. Sipariş boru hattı akışı yürütme depends_on.
comment str Akış açıklaması.
spark_conf dict Bu sorgunun yürütülmesi için Spark yapılandırmalarının listesi
import_checkpoint str Mevcut bir Yapılandırılmış Akış kontrol noktasına giden yol, böylece taşınan akış, kaynağı yeniden işlemek yerine son taahhüt edilen ofsetinden devam eder. Bir kontrol noktası içe aktarmak Beta'da. Bkz. Yapılandırılmış Akış Kontrol Noktasını Taşımak.

Örnekler

from pyspark import pipelines as dp

# Create a sink for an external Delta table
dp.create_sink("my_sink", "delta", {"path": "/tmp/delta_sink"})

# Add an append flow to an external Delta table
@dp.append_flow(name = "flow", target = "my_sink")
def flowFunc():
  return <streaming-query>

# Add a backfill
@dp.append_flow(name = "backfill", target = "my_sink", once = True)
def backfillFlowFunc():
    return (
      spark.read
      .format("json")
      .load("/path/to/backfill/")
    )

# Create a Kafka sink
dp.create_sink(
  "my_kafka_sink",
  "kafka",
  {
    "kafka.bootstrap.servers": "host:port",
    "topic": "my_topic"
  }
)

# Add an append flow to a Kafka sink
@dp.append_flow(name = "flow", target = "my_kafka_sink")
def myFlow():
  return read_stream("xxx").select(F.to_json(F.struct("*")).alias("value"))

Yapılandırılmış Akış kontrol noktasını Göç Et

Important

Bir kontrol noktası içe aktarmak Beta'da.

Mevcut bir Yapılandırılmış Akış iş yükünü kaynağı yeniden işlemeden bir boru hattına taşımak için kullanılır import_checkpoint . Bunu checkpointLocation , bulut depolama, Unity Kataloğu hacmi veya DBFS yolu olan Structured Streaming sorgusunuza ayarlayın. İlk boru hattı güncellemesinde, akış o kontrol noktasını pipeline'ın yönetilen depolamasına klonlar. Akış, son bağlanmış ofsetten (örneğin toplamalar, deduplication anahtarları ve watermarklar gibi) olduğu gibi devam eder. Sonraki boru hattı güncellemeleri akışın klonlanmış kontrol noktasını kullanır; orijinal kontrol noktası değiştirilmez.

Akış, create_table veya bir sink ile oluşturulan yönetilen bir tabloyu hedeflemelidir.

Warning

Pipeline'ı çalıştırmadan önce orijinal Structured Streaming sorgusunu durdurun. Orijinal Yapılandırılmış Akış sorgusu aktarmadan sonra tekrar kullanılabilir, ancak kontrol noktası durumunu yönetmeniz ve boru hattı ile sorgu aynı anda aynı tabloya yazmadığından emin olmalısınız; bu da tekrarlanan veri üretebilir.

Structured Streaming sorgusunu bir boru hattı akışı olarak yeniden oluşturun ve sorgu yazdığı tabloya yazmaya devam eder ve kontrol noktasını içe aktarın:

from pyspark import pipelines as dp

# Use the existing table your Structured Streaming query writes to; create_table adopts it.
dp.create_table("streaming_table")

# Continue from the imported checkpoint instead of reprocessing the source.
@dp.append_flow(
  target = "streaming_table",
  import_checkpoint = "/Volumes/my_catalog/my_schema/checkpoints/my_stream",
)
def migrate():
  # The same source your original query read from.
  return spark.readStream.table("source_table")

Kontrol noktası yalnızca ilk pipeline güncellemesinde içe aktarılır. Sonraki güncellemeler import_checkpoint. Tam yenileme kontrol noktasını yeniden içe aktarmaz. Yeni, boş bir kontrol noktasından başlıyor ve kaynağı yeniden işliyor. Farklı bir kontrol noktası içe aktarmak için, hedef tablo için daha önce kullanılmamış bir akış adı kullanın; mevcut bir akış adının tekrar kullanılması ithalatı atlar.

Sınırlamalar

  • Son güncelleştirme: