pyspark parametrize

import%20marimo%0A%0A__generated_with%20%3D%20%220.13.7%22%0Aapp%20%3D%20marimo.App(width%3D%22medium%22)%0A%0A%0A%40app.cell(hide_code%3DTrue)%0Adef%20_()%3A%0A%20%20%20%20import%20marimo%20as%20mo%0A%0A%20%20%20%20return%20(mo%2C)%0A%0A%0A%40app.cell(hide_code%3DTrue)%0Adef%20_(mo)%3A%0A%20%20%20%20mo.md(r%22%22%22%23%23%20Set%20Up%22%22%22)%0A%20%20%20%20return%0A%0A%0A%40app.cell%0Adef%20_()%3A%0A%20%20%20%20from%20datetime%20import%20date%0A%0A%20%20%20%20import%20pandas%20as%20pd%0A%20%20%20%20from%20pyspark.sql%20import%20SparkSession%0A%0A%20%20%20%20spark%20%3D%20SparkSession.builder.getOrCreate()%0A%20%20%20%20return%20date%2C%20pd%2C%20spark%0A%0A%0A%40app.cell%0Adef%20_(date%2C%20pd%2C%20spark)%3A%0A%20%20%20%20%23%20Create%20a%20Spark%20DataFrame%0A%20%20%20%20item_price_pandas%20%3D%20pd.DataFrame(%0A%20%20%20%20%20%20%20%20%7B%0A%20%20%20%20%20%20%20%20%20%20%20%20%22item_id%22%3A%20%5B1%2C%202%2C%203%2C%204%5D%2C%0A%20%20%20%20%20%20%20%20%20%20%20%20%22price%22%3A%20%5B4%2C%202%2C%205%2C%201%5D%2C%0A%20%20%20%20%20%20%20%20%20%20%20%20%22transaction_date%22%3A%20%5B%0A%20%20%20%20%20%20%20%20%20%20%20%20%20%20%20%20date(2025%2C%201%2C%2015)%2C%0A%20%20%20%20%20%20%20%20%20%20%20%20%20%20%20%20date(2025%2C%202%2C%201)%2C%0A%20%20%20%20%20%20%20%20%20%20%20%20%20%20%20%20date(2025%2C%203%2C%2010)%2C%0A%20%20%20%20%20%20%20%20%20%20%20%20%20%20%20%20date(2025%2C%204%2C%2022)%2C%0A%20%20%20%20%20%20%20%20%20%20%20%20%5D%2C%0A%20%20%20%20%20%20%20%20%7D%0A%20%20%20%20)%0A%0A%20%20%20%20item_price%20%3D%20spark.createDataFrame(item_price_pandas)%0A%20%20%20%20item_price.show()%0A%20%20%20%20return%20(item_price%2C)%0A%0A%0A%40app.cell(hide_code%3DTrue)%0Adef%20_(mo)%3A%0A%20%20%20%20mo.md(r%22%22%22%23%23%20Traditional%20Query%20Approach%22%22%22)%0A%20%20%20%20return%0A%0A%0A%40app.cell%0Adef%20_(item_price%2C%20spark)%3A%0A%20%20%20%20item_price.createOrReplaceTempView(%22item_price_view%22)%0A%20%20%20%20transaction_date_str%20%3D%20%222025-02-15%22%0A%0A%20%20%20%20query_with_fstring%20%3D%20f%22%22%22SELECT%20*%0A%20%20%20%20FROM%20item_price_view%0A%20%20%20%20WHERE%20transaction_date%20%3E%20'%7Btransaction_date_str%7D'%0A%20%20%20%20%22%22%22%0A%0A%20%20%20%20spark.sql(query_with_fstring).show()%0A%20%20%20%20return%20(transaction_date_str%2C)%0A%0A%0A%40app.cell(hide_code%3DTrue)%0Adef%20_(mo)%3A%0A%20%20%20%20mo.md(r%22%22%22%23%23%20Parameterized%20Queries%20with%20PySpark%20Custom%20String%20Formatting%22%22%22)%0A%20%20%20%20return%0A%0A%0A%40app.cell%0Adef%20_(item_price%2C%20spark%2C%20transaction_date_str)%3A%0A%20%20%20%20parametrized_query%20%3D%20%22%22%22SELECT%20*%0A%20%20%20%20FROM%20%7Bitem_price%7D%0A%20%20%20%20WHERE%20transaction_date%20%3E%20%7Btransaction_date%7D%0A%20%20%20%20%22%22%22%0A%0A%20%20%20%20spark.sql(%0A%20%20%20%20%20%20%20%20parametrized_query%2C%20item_price%3Ditem_price%2C%20transaction_date%3Dtransaction_date_str%0A%20%20%20%20).show()%0A%20%20%20%20return%0A%0A%0A%40app.cell(hide_code%3DTrue)%0Adef%20_(mo)%3A%0A%20%20%20%20mo.md(r%22%22%22%23%23%20Parameterized%20Queries%20with%20Parameter%20Markers%22%22%22)%0A%20%20%20%20return%0A%0A%0A%40app.cell%0Adef%20_(date%2C%20item_price%2C%20spark)%3A%0A%20%20%20%20query_with_markers%20%3D%20%22%22%22SELECT%20*%0A%20%20%20%20FROM%20%7Bitem_price%7D%0A%20%20%20%20WHERE%20transaction_date%20%3E%20%3Atransaction_date%0A%20%20%20%20%22%22%22%0A%0A%20%20%20%20transaction_date%20%3D%20date(2025%2C%202%2C%2015)%0A%0A%20%20%20%20spark.sql(%0A%20%20%20%20%20%20%20%20query_with_markers%2C%0A%20%20%20%20%20%20%20%20item_price%3Ditem_price%2C%0A%20%20%20%20%20%20%20%20args%3D%7B%22transaction_date%22%3A%20transaction_date%7D%2C%0A%20%20%20%20).show()%0A%20%20%20%20return%20(query_with_markers%2C)%0A%0A%0A%40app.cell(hide_code%3DTrue)%0Adef%20_(mo)%3A%0A%20%20%20%20mo.md(r%22%22%22%23%23%20Make%20SQL%20Easier%20to%20Reuse%22%22%22)%0A%20%20%20%20return%0A%0A%0A%40app.cell%0Adef%20_(date%2C%20item_price%2C%20query_with_markers%2C%20spark)%3A%0A%20%20%20%20transaction_date_1%20%3D%20date(2025%2C%203%2C%209)%0A%0A%20%20%20%20spark.sql(%0A%20%20%20%20%20%20%20%20query_with_markers%2C%0A%20%20%20%20%20%20%20%20item_price%3Ditem_price%2C%0A%20%20%20%20%20%20%20%20args%3D%7B%22transaction_date%22%3A%20transaction_date_1%7D%2C%0A%20%20%20%20).show()%0A%20%20%20%20return%0A%0A%0A%40app.cell%0Adef%20_(date%2C%20item_price%2C%20query_with_markers%2C%20spark)%3A%0A%20%20%20%20transaction_date_2%20%3D%20date(2025%2C%203%2C%2015)%0A%0A%20%20%20%20spark.sql(%0A%20%20%20%20%20%20%20%20query_with_markers%2C%0A%20%20%20%20%20%20%20%20item_price%3Ditem_price%2C%0A%20%20%20%20%20%20%20%20args%3D%7B%22transaction_date%22%3A%20transaction_date_2%7D%2C%0A%20%20%20%20).show()%0A%20%20%20%20return%0A%0A%0A%40app.cell(hide_code%3DTrue)%0Adef%20_(mo)%3A%0A%20%20%20%20mo.md(r%22%22%22%23%23%20Easier%20Unit%20Testing%20with%20Parameterized%20Queries%22%22%22)%0A%20%20%20%20return%0A%0A%0A%40app.cell%0Adef%20_(spark)%3A%0A%20%20%20%20def%20filter_by_price_threshold(df%2C%20amount)%3A%0A%20%20%20%20%20%20%20%20return%20spark.sql(%0A%20%20%20%20%20%20%20%20%20%20%20%20%22SELECT%20*%20from%20%7Bdf%7D%20where%20price%20%3E%20%3Aamount%22%2C%20df%3Ddf%2C%20args%3D%7B%22amount%22%3A%20amount%7D%0A%20%20%20%20%20%20%20%20)%0A%0A%20%20%20%20return%20(filter_by_price_threshold%2C)%0A%0A%0A%40app.cell%0Adef%20test_query_return_correct_number_of_rows(filter_by_price_threshold%2C%20spark)%3A%0A%20%20%20%20%23%20Create%20test%20input%20DataFrame%0A%20%20%20%20df%20%3D%20spark.createDataFrame(%0A%20%20%20%20%20%20%20%20%5B%0A%20%20%20%20%20%20%20%20%20%20%20%20(%22Product%201%22%2C%2010.0%2C%205)%2C%0A%20%20%20%20%20%20%20%20%20%20%20%20(%22Product%202%22%2C%2015.0%2C%203)%2C%0A%20%20%20%20%20%20%20%20%20%20%20%20(%22Product%203%22%2C%208.0%2C%202)%2C%0A%20%20%20%20%20%20%20%20%5D%2C%0A%20%20%20%20%20%20%20%20%5B%22name%22%2C%20%22price%22%2C%20%22quantity%22%5D%2C%0A%20%20%20%20)%0A%0A%20%20%20%20%23%20Execute%20query%20with%20parameters%0A%20%20%20%20assert%20filter_by_price_threshold(df%2C%2010).count()%20%3D%3D%201%0A%20%20%20%20assert%20filter_by_price_threshold(df%2C%208).count()%20%3D%3D%202%0A%20%20%20%20return%0A%0A%0A%40app.cell(hide_code%3DTrue)%0Adef%20imports()%3A%0A%20%20%20%20import%20pytest%0A%0A%20%20%20%20return%0A%0A%0Aif%20__name__%20%3D%3D%20%22__main__%22%3A%0A%20%20%20%20app.run()%0A