Equivalent for Airflow templating via Parameters?

[bogdanstate]

A very common use case in Airflow / Airflow-like system is the use of templating.

One typical use case: let's say I have a BigQueryTask that looks like this:

task = BigQueryTask(
    query="""
    SELECT [columns]
    FROM `fh-bigquery.reddit_comments.2012`
    """,
    [...]
)

I would like to parameterize this query on the source table, so that I can run my flow for new source tables that show up in this dataset every month / every year, as the BigQuery Reddit comments dataset grows (this is just an example, but is typical of any event log dataset).

I tried defining a Parameter called source_table, and calling format on it like so:

"""
SELECT [columns]
FROM `fh-bigquery.reddit_comments.{source_table}`
""".format(source_table=source_table)

This, of course, doesn't work. It just resolves to:

SELECT [columns]
FROM `fh-bigquery.reddit_comments.<Parameter: source_table>`

So I know I am doing something wrong. In Airflow I could handle this behavior via a parameter.

So far the most relevant part of the documentation seems to be the X-Files episode tutorial, but I can't quite understand how to resolve a parameter inside an argument to a task like query for BigQueryTask.

[bogdanstate]

I should mention I've also tried using:

        query_params=[
            ('source_table', "STRING", source_table),
            ('ds', "STRING", ds),
        ],

where source_table and ds are Parameters. This also results in an error:

TypeError: Object of type Parameter is not JSON serializable

[bogdanstate]

So, I think I answered my own question: having a "factory" function get_flow of the form:

def get_flow(ds, source_table):
    with Flow([...]) as flow:
          [create flow using ds and source_table] 
    return flow

Calling get_flow(ds='2014-01-01', source_table='2014') seems to work well, without requiring parameters at all!

[cicdw]

Hi @bogdanstate - remember that all Prefect tasks represent deferred computation. So when you format your query using a Parameter object (which has not run and produced a value yet), it's expected that you get a weird result. To properly achieve your goal, you need to place all this logic within the Flow so that it executes using runtime information:

source_table = Parameter("source_table")
bigquery = BigQueryTask([...]) # all init kwargs but NOT the query kwarg as it is not known yet


@task
def construct_query(source_table):
    return """
SELECT [columns]
FROM `fh-bigquery.reddit_comments.{source_table}`
""".format(source_table=source_table)


with Flow("BigQuery-template") as flow:
    query = construct_query(source_table)
    bigquery(query=query) # pass the runtime resolved value


flow.run(source_table="A")
flow.run(source_table="B")

There are also some string formatting tasks that help you with some of this boilerplate.

[bogdanstate]

Oh, I see, so the Parameter has to be defined outside of the flow context. I think I get it now.

[cicdw]

Actually that's irrelevant - the main point I was trying to highlight is that tasks are deferred and can't produce outputs until the Flow is running. Hence the need for the query constructor task that only runs at flow runtime and actually has access to the value of the Parameter.