Skip to content

Commit 6c994c3

Browse files
saumitra-rawatSaumitra Rawat (from Dev Box)Copilot
authored
Add DataExploration notebook (#7)
Co-authored-by: Saumitra Rawat (from Dev Box) <saurawat@microsoft.com> Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
1 parent f6ff9be commit 6c994c3

1 file changed

Lines changed: 154 additions & 0 deletions

File tree

‎source/resources/notebooks/DataExploration.ipynb‎

Lines changed: 154 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,154 @@
1+
{
2+
"cells": [
3+
{
4+
"cell_type": "code",
5+
"execution_count": null,
6+
"metadata": {},
7+
"outputs": [],
8+
"source": [
9+
"import org.apache.spark.sql.{DataFrame, SparkSession}\n",
10+
"import org.apache.spark.sql.expressions.Window\n",
11+
"import org.apache.spark.sql.functions._\n",
12+
"import org.apache.spark.sql.types._\n",
13+
"\n",
14+
"object DataExploration {\n",
15+
"\n",
16+
" // Configuration case class to hold settings\n",
17+
" case class ExplorationConfig(\n",
18+
" manifestFilePath: String,\n",
19+
" dataProductName: String,\n",
20+
" datasetName: String,\n",
21+
" datasetVersion: String\n",
22+
" )\n",
23+
"\n",
24+
" /**\n",
25+
" * Retrieves the full path to the IndexLatest.csv file based on the manifest configuration.\n",
26+
" */\n",
27+
" def getIndexFilePath(spark: SparkSession, config: ExplorationConfig): String = {\n",
28+
" import spark.implicits._\n",
29+
"\n",
30+
" val dfManifest = spark.read.option(\"multiline\", \"true\").json(config.manifestFilePath)\n",
31+
"\n",
32+
" val datasetManifestEntryDf =\n",
33+
" dfManifest\n",
34+
" .select(explode($\"data_products\").as(\"dp\"))\n",
35+
" .select(\n",
36+
" $\"dp.name\".as(\"dataProductName\"),\n",
37+
" coalesce($\"dp.enabled\", lit(true)).as(\"isDataProductEnabled\"),\n",
38+
" $\"dp.output_configuration.storage_account\".as(\"storage_account\"),\n",
39+
" $\"dp.output_configuration.container\".as(\"container\"),\n",
40+
" $\"dp.output_configuration.base_path\".as(\"base_path\"),\n",
41+
" explode($\"dp.tables\").as(\"tables\")\n",
42+
" )\n",
43+
" .filter(\n",
44+
" $\"isDataProductEnabled\" &&\n",
45+
" $\"dataProductName\" === config.dataProductName &&\n",
46+
" $\"tables.name\" === config.datasetName &&\n",
47+
" $\"tables.version\".cast(\"string\") === config.datasetVersion\n",
48+
" )\n",
49+
"\n",
50+
" if (datasetManifestEntryDf.head(1).isEmpty) {\n",
51+
" mssparkutils.notebook.exit(\n",
52+
" s\"Manifest file is empty/unreadable: ${config.manifestFilePath} or ${config.dataProductName} is disabled for copy.\"\n",
53+
" )\n",
54+
" }\n",
55+
"\n",
56+
" datasetManifestEntryDf\n",
57+
" .withColumn(\n",
58+
" \"fullPath\",\n",
59+
" concat(\n",
60+
" lit(\"abfss://\"),\n",
61+
" $\"container\",\n",
62+
" lit(\"@\"),\n",
63+
" $\"storage_account\",\n",
64+
" lit(\".dfs.<storageEnvironmentSuffix>\"),\n",
65+
" $\"base_path\",\n",
66+
" lit(config.datasetName),\n",
67+
" coalesce($\"tables.sub_path\", lit(\"\")),\n",
68+
" lit(\"/index-latest.csv\")\n",
69+
" )\n",
70+
" )\n",
71+
" .select($\"fullPath\")\n",
72+
" .as[String]\n",
73+
" .head()\n",
74+
" }\n",
75+
"\n",
76+
" /**\n",
77+
" * Reads the index file and returns the list of paths for the latest snapshot.\n",
78+
" */\n",
79+
" def getSnapshotPaths(spark: SparkSession, indexFilePath: String): Array[String] = {\n",
80+
" import spark.implicits._\n",
81+
" \n",
82+
" val window = Window.partitionBy(\"Partition\").orderBy(col(\"RunId\").desc)\n",
83+
" println(indexFilePath)\n",
84+
"\n",
85+
" spark.read\n",
86+
" .option(\"header\", \"true\")\n",
87+
" .csv(indexFilePath)\n",
88+
" .filter(col(\"IngestionMode\") === \"Snapshot\")\n",
89+
" .withColumn(\"rnk\", rank().over(window))\n",
90+
" .filter(col(\"rnk\") === 1)\n",
91+
" .select(\"Path\").distinct.as[String].collect()\n",
92+
" }\n",
93+
"\n",
94+
" /**\n",
95+
" * Loads the data from the provided paths into a DataFrame.\n",
96+
" */\n",
97+
" def loadData(spark: SparkSession, paths: Array[String]): DataFrame = {\n",
98+
" if (paths.isEmpty) {\n",
99+
" spark.emptyDataFrame\n",
100+
" } else {\n",
101+
" spark.read.parquet(paths: _*)\n",
102+
" }\n",
103+
" }\n",
104+
"\n",
105+
" /**\n",
106+
" * Helper method to execute the flow with the provided configuration.\n",
107+
" */\n",
108+
" def run(spark: SparkSession, config: ExplorationConfig): DataFrame = {\n",
109+
" val indexFilePath = getIndexFilePath(spark, config)\n",
110+
" val paths = getSnapshotPaths(spark, indexFilePath)\n",
111+
" loadData(spark, paths)\n",
112+
" }\n",
113+
"}"
114+
]
115+
},
116+
{
117+
"cell_type": "code",
118+
"execution_count": null,
119+
"metadata": {},
120+
"outputs": [],
121+
"source": [
122+
"// // Configure below parameters based on your data requirements\n",
123+
"// val config = DataExploration.ExplorationConfig(\n",
124+
"// manifestFilePath = \"abfss://config@<storageAccountName>.dfs.<storageEnvironmentSuffix>/manifest/v1.0/subscription-catalog.json\",\n",
125+
"// dataProductName = \"Products\",\n",
126+
"// datasetName = \"CDM_Products\",\n",
127+
"// datasetVersion = \"3.0\"\n",
128+
"// )\n",
129+
"// val df = DataExploration.run(spark, config)"
130+
]
131+
}
132+
],
133+
"metadata": {
134+
"description": null,
135+
"kernel_info": {
136+
"name": "synapse_pyspark"
137+
},
138+
"kernelspec": {
139+
"display_name": "Synapse PySpark",
140+
"language": "Python",
141+
"name": "synapse_pyspark"
142+
},
143+
"language_info": {
144+
"name": "scala"
145+
},
146+
"save_output": true,
147+
"synapse_widget": {
148+
"state": {},
149+
"version": "0.1"
150+
}
151+
},
152+
"nbformat": 4,
153+
"nbformat_minor": 2
154+
}

0 commit comments

Comments
 (0)