用于Read_Stream联合的数据库DLT语法

Question

对于DLT来说，很难用python语法通过dlt.read_stream操作符作为另外两个活动表的联合(unionByName)返回数据。

我的管道如下..。

作品: Table1：

@dlt.table()
def table_1()
   return spark.sql (''' select mergeKey, date_seq, colN, case/when.., cast.. from live.raw_table_1 ''')

作品: Table2：

@dlt.table()
def table_2()
   return spark.sql (''' select mergeKey, date_seq, colN, case/when.., cast.. from live.raw_table_2 ''')

问题: View_Union：

    @dlt.view()
    def pre_merge_union_v
    --> question: what's the syntax to [return dlt.read_stream("df_unioned").as.table1.unionByName(table2)

创建银表：

dlt.create_streaming_live_table("silver_table")

最后，将更改应用于Silver：

dlt.apply_changes(
   target = "silver_table"),
   source = "pre_merge_union_v"
   keys = ["mergeKey"],
   sequence_by = "date_seq"

试过：

我尝试将我的View_Union视图创建为：

@dlt.view()
def pre_merge_union_v:
   df_table1 = spark.table("live.raw_table1")
   df_table2 = spark.table("live.raw_table2")
   df_unioned = df_table1.unionByName(df_table2)
     return df_unioned

然而，当管道运行时，它会抱怨pre_merge_union_v必须是一个READ_STREAM视图(因为它正在合并到一个流直播表中，我想)

使用return dlt.read_stream(df_unioned)还会生成一个错误。

Answer 1

我想我已经回答了我自己的问题。希望能帮上忙。

return dlt.read_stream("raw_table1").unionByName(dlt.read_stream("raw_table2"))