spark/scala/iceberg_delta_spark_example.ipynb

{"nbformat_minor": 2, "cells": [{"source": "# Example Objective\n\n Environment: HDI 5.0 (Spark 3.x.x)\n\nThe objective of this example code is to demonstrate how you can use Spark, Iceberg, and Delta together:\n\n- Store master data into Spark table default to parquet format\n- Store sales transactions data in the Delta table \n- Store return transactions data in the Iceberg table\n\nAnd finally, join these three tables to query some business insight. \n\n# Configuration\n\n- Add Delta Lake Connector Libraries\n- Add Iceberg Connector runtime jar\n- Add Iceberg and Delta Lake SQL extensions\n\nIn this example, we are using the Hadoop warehouse as a catalog for the Iceberg, and the catalog name is `demo_iceberg`. The base path for the warehouse directory is configured by `spark.sql.catalog.iceberg.warehouse`\n\nThe Delta catalog adds support for Delta tables to Spark\u2019s built-in catalog (using spark.sql.catalog.spark_catalog), and delegates to the built-in catalog for non-Delta tables. Spark default warehouse location is your primary container `/apps/spark/warehouse`. Delta data also will be stored at Spark's default warehouse location.", "cell_type": "markdown", "metadata": {"editable": true, "deletable": true}}, {"execution_count": null, "cell_type": "code", "source": "%%configure -f\n{ \"conf\": {\"spark.jars.packages\": \"org.apache.iceberg:iceberg-spark-runtime-3.1_2.12:1.1.0,io.delta:delta-core_2.12:1.0.1,net.andreinc:mockneat:0.4.8\",\n           \"spark.sql.extensions\":\"org.apache.iceberg.spark.extensions.IcebergSparkSessionExtensions,io.delta.sql.DeltaSparkSessionExtension\",\n           \"spark.sql.catalog.spark_catalog\":\"org.apache.spark.sql.delta.catalog.DeltaCatalog\",\n           \"spark.sql.catalog.demo_iceberg\":\"org.apache.iceberg.spark.SparkCatalog\",\n           \"spark.sql.catalog.demo_iceberg.type\":\"hadoop\",\n           \"spark.sql.catalog.demo_iceberg.warehouse\":\"/iceberg/warehouse\"\n          }\n}", "outputs": [], "metadata": {"cell_status": {"execute_time": {"duration": 31.43310546875, "end_time": 1677518579493.564}}, "editable": true, "collapsed": false, "deletable": true}}, {"source": "## Model Objects\n- Product - Product Master information - Store this information in Spark Table\n- Product sales transactions- Product sales quantity and transaction date - Store this information in Delta Table\n- Product return transactions - Product return quantity and transaction date - Store this information in Iceberg Table", "cell_type": "markdown", "metadata": {"editable": true, "deletable": true}}, {"execution_count": null, "cell_type": "code", "source": "// product model\ncase class Product(id:Int, name: String, price:Float)\n//product sales\ncase class ProductSales(productId:Int, soldQty:Int, saleDate:String)\n//product return model\ncase class ProductReturn(productId:Int, returnQty:Int, returnDate:String)", "outputs": [], "metadata": {"cell_status": {"execute_time": {"duration": 2487.522216796875, "end_time": 1677518629838.673}}, "editable": true, "collapsed": false, "deletable": true}}, {"source": "# Generate Master Data\n- Customer\n- Product", "cell_type": "markdown", "metadata": {"collapsed": true, "editable": true, "deletable": true}}, {"execution_count": null, "cell_type": "code", "source": "import net.andreinc.mockneat.MockNeat\nimport net.andreinc.mockneat.abstraction.MockUnit\nimport net.andreinc.mockneat.types.enums.RandomType\nimport java.text.DecimalFormat\nimport java.time.LocalDate\n\n//configure base on your need\n// this program will run on driver side limit by driver memory\nval dateStart = LocalDate.of(2014, 1, 1)\nval dateEnd = LocalDate.of(2015, 1, 1)\nval numProducts=10\nval numOfSalesTx = 10000\nval numOfReturnTx = 1000\n// constants ends\n\nval df = new DecimalFormat(\"0.00\");\nval mockNeat = MockNeat.threadLocal()\n\nval productData = (1 to numProducts).map(i=>{\n    Product(i,s\"Product${i}\",df.format(mockNeat.floats().range(10.0f, 20.0f).get()).toFloat)\n})", "outputs": [], "metadata": {"cell_status": {"execute_time": {"duration": 2321.671875, "end_time": 1677521855658.605}}, "editable": true, "collapsed": false, "deletable": true}}, {"source": "## Generate Transaction Data\n- Product sales transactions\n- Product return transactions", "cell_type": "markdown", "metadata": {"cell_status": {"execute_time": {"duration": 785.129150390625, "end_time": 1677177083931.973}}, "editable": true, "deletable": true}}, {"execution_count": null, "cell_type": "code", "source": "import scala.collection.JavaConverters._\n\n//index MockUnit for Product\nval prodIndex = mockNeat.ints().range(0,numProducts-1)\n\nval productSalesData = (1 to numOfSalesTx).map(i=>{\n    ProductSales(productData(prodIndex.get()).id, mockNeat.ints().range(1,5).get(),\n                 mockNeat.localDates.between(dateStart, dateEnd).mapToString().get())\n})\n\nval productReturnData = (1 to numOfReturnTx).map(i=>{\n    ProductReturn(productData(prodIndex.get()).id, mockNeat.ints().range(1,5).get(),\n                  mockNeat.localDates.between(dateStart, dateEnd).mapToString().get())\n})\n", "outputs": [], "metadata": {"cell_status": {"execute_time": {"duration": 1277.9091796875, "end_time": 1677521860573.661}}, "editable": true, "collapsed": false, "deletable": true}}, {"source": "# Save Data\n- Product - Store this information in Spark Table\n- Product Sales Transactions -Store this information in Delta Table\n- Product Return Transactions -Store this information in Iceberg Table", "cell_type": "markdown", "metadata": {"editable": true, "deletable": true}}, {"execution_count": null, "cell_type": "code", "source": "sc.parallelize(productData).toDF.write.mode(\"overwrite\").saveAsTable(\"product\")\nsc.parallelize(productSalesData).toDF.write.format(\"delta\").mode(\"overwrite\").saveAsTable(\"salestx\")\nsc.parallelize(productReturnData).toDF.write.format(\"iceberg\").mode(\"overwrite\").saveAsTable(\"demo_iceberg.returntx\")", "outputs": [], "metadata": {"cell_status": {"execute_time": {"duration": 9322.535888671875, "end_time": 1677521874402.556}}, "editable": true, "collapsed": false, "deletable": true}}, {"source": "## Iceberg Table Metadata Log Entries ", "cell_type": "markdown", "metadata": {}}, {"execution_count": null, "cell_type": "code", "source": "%%sql\nselect * from demo_iceberg.returntx.metadata_log_entries", "outputs": [], "metadata": {"cell_status": {"execute_time": {"duration": 770.799072265625, "end_time": 1677521879352.45}}, "editable": true, "collapsed": false, "deletable": true}}, {"source": "## Iceberg Table History and Application ID", "cell_type": "markdown", "metadata": {}}, {"execution_count": null, "cell_type": "code", "source": "%%sql\nselect\n    h.made_current_at,\n    s.operation,\n    h.snapshot_id,\n    h.is_current_ancestor,\n    s.summary['spark.app.id']\nfrom demo_iceberg.returntx.history h\njoin demo_iceberg.returntx.snapshots s\n  on h.snapshot_id = s.snapshot_id\norder by made_current_at", "outputs": [], "metadata": {"cell_status": {"execute_time": {"duration": 966.80908203125, "end_time": 1677521883617.476}}, "collapsed": false}}, {"source": "## Iceberg Data Files", "cell_type": "markdown", "metadata": {}}, {"execution_count": null, "cell_type": "code", "source": "%%sql\nSELECT * FROM demo_iceberg.returntx.files", "outputs": [], "metadata": {"cell_status": {"execute_time": {"duration": 1306.97900390625, "end_time": 1677521892153.774}}, "collapsed": false}}, {"source": "# Read Delta Sales and Iceberg Return Transactions\n- Read sales transactions from Delta Table\n- Read return transactions from Iceberg Table\n- Read product master from Spark Parquet", "cell_type": "markdown", "metadata": {}}, {"execution_count": null, "cell_type": "code", "source": "val salesTxDF = spark.sql(\"SELECT * FROM salestx\")\nval returnTxDF = spark.sql(\"SELECT * FROM demo_iceberg.returntx\")\nval productDF = spark.sql(\"SELECT * FROM product\")\n// join sales and return Tx DataFarme\n\nval salesReturnTxDF = salesTxDF.join(returnTxDF,salesTxDF(\"saleDate\")===returnTxDF(\"returnDate\") &&  salesTxDF(\"productId\")===returnTxDF(\"productId\"),\"fullouter\").\n    withColumn(\"date\",when(col(\"saleDate\")===null,col(\"returnDate\")).otherwise(col(\"saleDate\"))).\n        select(\"salestx.productId\",\"date\",\"soldQty\",\"returnQty\").\n            na.fill(0).\n            withColumn(\"absSoldQty\",col(\"soldQty\")-col(\"returnQty\"))\n// join with product\nval producSaleReturnTx = salesReturnTxDF.\n                            join(productDF,salesReturnTxDF(\"productId\")===productDF(\"id\"),\"inner\").\n                                withColumn(\"salesDollar\",col(\"absSoldQty\")*col(\"price\"))\nproducSaleReturnTx.registerTempTable(\"prodsalereturn\")", "outputs": [], "metadata": {"cell_status": {"execute_time": {"duration": 2290.659912109375, "end_time": 1677522585268.474}}, "editable": true, "collapsed": false, "deletable": true}}, {"source": "## Summary By Product Sold Qty", "cell_type": "markdown", "metadata": {"cell_status": {"execute_time": {"duration": 2328.51904296875, "end_time": 1677521905055.248}}}}, {"execution_count": null, "cell_type": "code", "source": "%%sql\nSELECT name, Sum(absSoldQty) FROM prodsalereturn GROUP By name", "outputs": [], "metadata": {"cell_status": {"execute_time": {"duration": 7358.261962890625, "end_time": 1677522366441.366}}, "collapsed": false}}, {"source": "## Summary By Dollar Sales By Date", "cell_type": "markdown", "metadata": {}}, {"execution_count": null, "cell_type": "code", "source": "%%sql\nSELECT date, Sum(salesDollar) FROM prodsalereturn GROUP By date", "outputs": [], "metadata": {"cell_status": {"execute_time": {"duration": 7381.121826171875, "end_time": 1677522671924.796}}, "collapsed": false}}, {"execution_count": null, "cell_type": "code", "source": "", "outputs": [], "metadata": {"collapsed": true}}], "nbformat": 4, "metadata": {"kernelspec": {"display_name": "Spark", "name": "sparkkernel", "language": ""}, "language_info": {"mimetype": "text/x-scala", "pygments_lexer": "scala", "name": "scala", "codemirror_mode": "text/x-scala"}}}