Lakeflow SDP Append Flow

[IM_01]

Hi All,
I'm using  append_flow to ingest data into the target table. Before returning the dataframe, I compute few column trnasformations, but those values aren't being calculated correctly (or: aren't showing up at all)
so the append flow should not include any column transformations.Does it need to include only reading source table?

[Ashwin_DSA]

Hi @IM_01,

append_flow is not limited to just reading a source table. You can (and should) include column transformations, filters, and any other DataFrame operations inside the function decorated with @dp.append_flow. The function simply needs to return a valid streaming DataFrame, and whatever transformations you apply to that DataFrame before returning it will be reflected in the data written to the target table.

Here's a quick example showing transformations inside an append_flow:

from pyspark import pipelines as dp
from pyspark.sql.functions import col, upper, current_timestamp

dp.create_streaming_table("customers_silver")

@dp.append_flow(target="customers_silver")
def customers_flow():
    return (
        spark.readStream.table("customers_bronze")
        .withColumn("name_upper", upper(col("name")))
        .withColumn("ingested_at", current_timestamp())
        .select("id", "name_upper", "region", "ingested_at")
    )

This is functionally identical to defining the transformations in a default flow with@dp.table(). The  confirms that the decorated function simply needs to return a streaming DataFrame from a "user-defined query". There's no restriction against transformations.

Your transformations may not be showing up because of various reasons as listed below

1. If you created the streaming table with an explicit schema using create_streaming_table(name, schema=...), the target table's schema takes precedence. Columns returned by your flow that don't exist in the target schema are silently dropped, and columns in the target schema not returned by your flow come through as NULL. Make sure your transformed column names match the target table's schema exactly.
   
   
2. A row that has already been appended to a streaming table will not be re-queried with later updates. If you added or changed your transformations after data was already ingested, the existing rows won't reflect the new logic...  only newly arriving rows will. You would need to trigger a  to reprocess all historical data with the updated transformations.
   
   
3. If you are using the SQL interface (CREATE FLOW ... INSERT INTO target BY NAME), columns are matched by name, not position. Any column in your SELECT that doesn't have a matching name in the target table is ignored.

Rather than stripping transformations out of your append_flow, keep them there... that's the intended pattern. To troubleshoot, try:

1. Check whether your target table was created with an explicit schema, and verify the column names and types match what your flow returns.
   
   
2. If you recently changed your transformation logic, run a full refresh to reprocess existing data.
   
   
3. Add a temporary display() or logging step in your notebook (outside the pipeline) to inspect the DataFrame your flow function returns, confirming the transformations produce the expected output before the data reaches the target.

Hope this helps.

If this answer resolves your question, could you mark it as “Accept as Solution”? That helps other users quickly find the correct fix.