append_flow

Dekorator @dp.append_flow membuat saluran lampiran atau pengisian ulang untuk tabel saluran Anda. Fungsi harus mengembalikan DataFrame streaming Apache Spark. Lihat Memuat dan memproses data secara bertahap dengan alur Lakeflow.

Alur tambahan dapat menargetkan tabel streaming, tabel terkelola, atau sink.

Syntax

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

Parameter-parameternya

Pengaturan	Tipe	Description
fungsi	function	Dibutuhkan. Fungsi yang mengembalikan DataFrame streaming Apache Spark dari kueri yang ditentukan pengguna.
target	str	Dibutuhkan. Nama tabel atau sink yang merupakan target alur penambahan.
name	str	Nama alur. Jika tidak disediakan, akan otomatis menggunakan nama fungsi.
once	bool	Secara opsional, tentukan alur sebagai alur satu kali, seperti isi ulang. Dengan menggunakan once=True, alurnya dapat diubah dengan dua cara: Nilai yang dikembalikan. streaming-query. harus berupa DataFrame batch dalam hal ini, bukan DataFrame streaming. Alur dijalankan satu kali secara default. Jika alur diperbarui dengan pembaruan lengkap, maka ONCE alur akan berjalan kembali untuk membuat ulang data.
depends_on	str atau list	Pratinjau Publik Satu atau lebih nama alur yang harus berhasil diselesaikan sebelum alur ini dimulai. Menerima satu nama alur atau daftar nama. Ini hanya mengatur eksekusi alur; ini tidak mengubah cara alur berjalan. Lihat Eksekusi aliran pipeline pesanan dengan depends_on.
comment	str	Deskripsi untuk alur.
spark_conf	dict	Daftar konfigurasi Spark untuk eksekusi kueri ini
import_checkpoint	str	Jalur ke checkpoint Structured Streaming yang sudah ada untuk diimpor ke alur, sehingga aliran yang dimigrasikan dilanjutkan dari offset terakhir yang dikomitmen, bukan memproses ulang sumbernya. Mengimpor checkpoint masih dalam tahap Beta. Lihat Migrasi checkpoint Streaming Terstruktur.

Examples

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

Migrasi checkpoint Streaming Terstruktur

Important

Mengimpor checkpoint masih dalam tahap Beta.

Digunakan import_checkpoint untuk memigrasikan beban kerja Structured Streaming yang sudah ada ke pipeline tanpa memproses ulang sumbernya. Atur ke query Structured Streaming yang checkpointLocation Anda gunakan, yang bisa berupa penyimpanan cloud, volume Unity Catalog, atau jalur DBFS. Pada pembaruan pipeline pertama, flow mengkloning checkpoint tersebut ke penyimpanan yang dikelola pipeline. Aliran kemudian dilanjutkan dari offset terakhir yang dikomitmenkan dengan statusnya (seperti agregasi, kunci deduplikasi, dan watermark) tetap utuh. Pembaruan pipeline berikutnya menggunakan checkpoint kloning aliran; checkpoint asli tidak dimodifikasi.

Aliran harus menargetkan tabel yang dikelola yang dibuat dengan create_table atau sink.

Hentikan kueri Structured Streaming asli sebelum Anda menjalankan pipeline. Kueri Structured Streaming asli dapat digunakan kembali setelah impor, tetapi Anda perlu mengelola status checkpoint-nya dan memastikan pipeline serta kueri tidak menulis ke tabel yang sama pada waktu yang sama, yang dapat menghasilkan data duplikat.

Buat ulang kueri Structured Streaming sebagai alur pipeline yang menulis ke tabel baru dan mengimpor checkpoint-nya:

from pyspark import pipelines as dp

# Create a new managed table for the pipeline
dp.create_table("target_table")

# Continue from the imported checkpoint instead of reprocessing the source.
@dp.append_flow(
  target = "target_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")

Checkpoint hanya diimpor sekali, pada pembaruan pipeline pertama; pembaruan selanjutnya mengabaikan import_checkpoint. Penyegaran penuh tidak mengimpor ulang checkpoint; checkpoint dimulai dari checkpoint baru yang kosong dan memproses ulang sumbernya. Untuk mengimpor checkpoint yang berbeda, gunakan nama alur yang belum pernah digunakan untuk tabel target sebelumnya; menggunakan kembali nama alur yang sudah ada akan melewati impor.

Keterbatasan

• Mengimpor checkpoint ke dalam tabel yang sudah ada (misalnya, target kueri Structured Streaming asli) tidak didukung. Targetkan tabel baru yang dibuat pipeline, atau sink.
• import_checkpoint hanya didukung pada append_flow.