Skip to content

Commit 6609e65

Browse files
shivsoodshivsood
committed
stream to external tables using MsSQLSpark connector
1 parent a624a39 commit 6609e65

1 file changed

Lines changed: 100 additions & 0 deletions

File tree

‎samples/features/sql-big-data-cluster/spark/data-virtualization/MSSQLSparkConnector_StreamingToDataPools.ipynb‎

Lines changed: 100 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,100 @@
1+
{
2+
"metadata": {
3+
"kernelspec": {
4+
"name": "sparkkernel",
5+
"display_name": "Spark | Scala"
6+
},
7+
"language_info": {
8+
"name": "scala",
9+
"mimetype": "text/x-scala",
10+
"codemirror_mode": "text/x-scala",
11+
"pygments_lexer": "scala"
12+
}
13+
},
14+
"nbformat_minor": 2,
15+
"nbformat": 4,
16+
"cells": [
17+
{
18+
"cell_type": "markdown",
19+
"source": "# Stream to datapool using the MSSQL Spark Connector\r\nMSSQL Spark connector provides an efficient write to SQLServer master instance and SQL Server data pool in Big Data Clusters. This sample shows how to use the connector to stream a file to an external table in data pools.\r\n\r\n The sample is divided into 2 parts. \r\n- In Part 1, we stream all files under a given directory to a external table in data pools\r\n- In Part 2, we read the table to see the data. \r\n\r\nPreReq: \r\n- The sample uses a SQL database named \"MyTestDatabase\". Create this before you run this sample. The database can be created as follows\r\n ``` sql\r\n Create DATABASE MyTestDatabase\r\n GO \r\n ``` \r\n- Download [AdultCensusIncome.csv]( https://amldockerdatasets.azureedge.net/AdultCensusIncome.csv ) to your local machine. Create a hdfs folder named /filestreaming and upload the files there. \r\n",
20+
"metadata": {}
21+
},
22+
{
23+
"cell_type": "markdown",
24+
"source": "## Part 1 - Stream all files under a given directory to an external table in data pools\r\n ",
25+
"metadata": {}
26+
},
27+
{
28+
"cell_type": "markdown",
29+
"source": "### Configure the user, password, database, sourceDir, datapool_tables,schema, datasource_name per your needs\r\n\r\n\r\n",
30+
"metadata": {}
31+
},
32+
{
33+
"cell_type": "code",
34+
"source": "import org.apache.spark.sql.types._\r\n\r\n//Change per your installation\r\nval user=\"sa\"\r\nval password=\"****\"\r\nval database = \"MyTestDatabase\"\r\nval sourceDir = \"/file_streaming\"\r\nval datapool_table = \"streaming_DataPoolTable\"\r\nval datasource_name = \"test_data_src\"\r\nval schema = StructType(Seq(\r\n StructField(\"age\",StringType,true), StructField(\"workclass\",StringType,true), StructField(\"fnlwgt\",StringType,true), StructField(\"education\",StringType,true), \r\n StructField(\"education-num\",StringType,true), StructField(\"marital-status\",StringType,true), StructField(\"occupation\",StringType,true), \r\n StructField(\"relationship\",StringType,true), StructField(\"race\",StringType,true), StructField(\"sex\",StringType,true), StructField(\"capital-gain\",StringType,true), \r\n StructField(\"capital-loss\",StringType,true), StructField(\"hours-per-week\",StringType,true), StructField(\"native-country\",StringType,true), StructField(\"income\",StringType,true)\r\n))\r\n\r\nval hostname = \"master-0.master-svc\"\r\nval port = 1433\r\nval url = s\"jdbc:sqlserver://${hostname}:${port};database=${database};user=${user};password=${password};\"",
35+
"metadata": {},
36+
"outputs": [
37+
{
38+
"output_type": "stream",
39+
"name": "stdout",
40+
"text": "Starting Spark application\n"
41+
},
42+
{
43+
"output_type": "display_data",
44+
"data": {
45+
"text/plain": "<IPython.core.display.HTML object>",
46+
"text/html": "<table>\n<tr><th>ID</th><th>YARN Application ID</th><th>Kind</th><th>State</th><th>Spark UI</th><th>Driver log</th><th>Current session?</th></tr><tr><td>74</td><td>application_1568069140269_0144</td><td>spark</td><td>idle</td><td><a target=\"_blank\" href=\"https://sparkhead-0.sparkhead-svc:8090/proxy/application_1568069140269_0144/\">Link</a></td><td><a target=\"_blank\" href=\"https://10.91.90.171:30443/gateway/default/yarn/container/container_1568069140269_0144_01_000001/root\">Link</a></td><td>✔</td></tr></table>"
47+
},
48+
"metadata": {}
49+
},
50+
{
51+
"output_type": "stream",
52+
"name": "stdout",
53+
"text": "SparkSession available as 'spark'.\n"
54+
},
55+
{
56+
"output_type": "stream",
57+
"name": "stdout",
58+
"text": "import org.apache.spark.sql.types._\nuser: String = sa\npassword: String = Yukon900\ndatabase: String = MyTestDatabase\nsourceDir: String = /file_streaming\ndatapool_table: String = streaming_DataPoolTable\ndatasource_name: String = test_data_src\nschema: org.apache.spark.sql.types.StructType = StructType(StructField(age,StringType,true), StructField(workclass,StringType,true), StructField(fnlwgt,StringType,true), StructField(education,StringType,true), StructField(education-num,StringType,true), StructField(marital-status,StringType,true), StructField(occupation,StringType,true), StructField(relationship,StringType,true), StructField(race,StringType,true), StructField(sex,StringType,true), StructField(capital-gain,StringType,true), StructField(capital-loss,StringType,true), StructField(hours-per-week,StringType,true), StructField(native-country,StringType,true), StructField(income,StringType,true))\nhostname: String = master-0.master-svc\nport: Int = 1433\nurl: String = jdbc:sqlserver://master-0.master-svc:1433;database=MyTestDatabase;user=sa;password=Yukon900;\n"
59+
}
60+
],
61+
"execution_count": 3
62+
},
63+
{
64+
"cell_type": "markdown",
65+
"source": " ### Run Spark Ingestion Job\r\n\r\nThis spark ingestion job is created using a **readStream** and a **writeStream** ",
66+
"metadata": {}
67+
},
68+
{
69+
"cell_type": "code",
70+
"source": " import org.apache.spark.sql.{SparkSession, SaveMode, Row, DataFrame}\r\n\r\nval df = spark.readStream.format(\"csv\").schema(schema).option(\"header\", true).load(sourceDir)\r\nval query = df.writeStream.outputMode(\"append\").foreachBatch{ (batchDF: DataFrame, batchId: Long) => \r\n batchDF.write\r\n .format(\"com.microsoft.sqlserver.jdbc.spark\")\r\n .mode(\"append\")\r\n .option(\"url\", url)\r\n .option(\"dbtable\", datapool_table)\r\n .option(\"user\", user)\r\n .option(\"password\", password)\r\n .option(\"dataPoolDataSource\",datasource_name).save()\r\n }.start()\r\n\r\nquery.processAllAvailable()\r\nquery.awaitTermination(40000)\r\n",
71+
"metadata": {},
72+
"outputs": [
73+
{
74+
"output_type": "stream",
75+
"name": "stdout",
76+
"text": "import org.apache.spark.sql.{SparkSession, SaveMode, Row, DataFrame}\ndf: org.apache.spark.sql.DataFrame = [age: string, workclass: string ... 13 more fields]\nquery: org.apache.spark.sql.streaming.StreamingQuery = org.apache.spark.sql.execution.streaming.StreamingQueryWrapper@1033421d\nres14: Boolean = false\n"
77+
}
78+
],
79+
"execution_count": 8
80+
},
81+
{
82+
"cell_type": "markdown",
83+
"source": "## Part2 - Read the external table using MSSQLSpark Connector \r\nNow you are streaming data from the source directory to the data pool table. An external table has been created in the targeted database specified above. You can view the table in the explorer tree and query it using t-sql. \r\n\r\nIf you want to view the current count of records in the external table use the code below. The stream with append new data to the external table as well. Try creating another census csv file with 100 rows and add it to the source directory.",
84+
"metadata": {}
85+
},
86+
{
87+
"cell_type": "code",
88+
"source": "def df_read(dbtable: String,\r\n url: String,\r\n dataPoolDataSource: String=\"\"): DataFrame = {\r\n spark.read\r\n .format(\"com.microsoft.sqlserver.jdbc.spark\")\r\n .option(\"url\", url)\r\n .option(\"dbtable\", dbtable)\r\n .option(\"user\", user)\r\n .option(\"password\", password)\r\n .option(\"dataPoolDataSource\", dataPoolDataSource)\r\n .load()\r\n }\r\n\r\nval new_df = df_read(datapool_table, url, dataPoolDataSource=datasource_name)\r\nprintln(\"Number of rows is \" + new_df.count)",
89+
"metadata": {},
90+
"outputs": [
91+
{
92+
"output_type": "stream",
93+
"name": "stdout",
94+
"text": "df_read: (dbtable: String, url: String, dataPoolDataSource: String)org.apache.spark.sql.DataFrame\nnew_df: org.apache.spark.sql.DataFrame = [age: string, workclass: string ... 13 more fields]\nNumber of rows is 618665\n"
95+
}
96+
],
97+
"execution_count": 9
98+
}
99+
]
100+
}

0 commit comments

Comments
 (0)