mirror of
https://github.com/Microsoft/sql-server-samples.git
synced 2025-12-08 14:58:54 +00:00
Updated to MSSQL Spark connector sample, new samples for accessing spark via restful livy, configuring a spark session and installing 3rd party pacakgee
This commit is contained in:
ShivSood
@@ -2,7 +2,7 @@
|
||||
|
||||
SQL Server Big Data cluster bundles Spark and HDFS together with SQL server. Azure Data Studio IDE provides built in notebooks that enables data scientists and data engineers to run Spark notebooks and job in Python, R, or Scala code against the Big Data Cluster. This folder contains spark sample notebook on using Spark in SQL server Big data cluster
|
||||
|
||||
## Folder contents
|
||||
## Contents
|
||||
|
||||
[PySpark Hello World](dataloading/hello_PySpark.ipynb)
|
||||
|
||||
@@ -14,14 +14,18 @@ SQL Server Big Data cluster bundles Spark and HDFS together with SQL server. Azu
|
||||
|
||||
[Data Transfer - Spark to SQL using Spark JDBC connector](data-virtualization/spark_to_sql_jdbc.ipynb/)
|
||||
|
||||
[Data Transfer - Spark to SQL using MSSQL Spark connector](spark_to_sql/mssql_spark_connector.ipynb/)
|
||||
[Data Transfer - Spark to SQL using MSSQL Spark connector](data-virtualization/mssql_spark_connector.ipynb/)
|
||||
|
||||
[Configure - Configure a spark session using a notebook](config-install/configure_spark_session.ipynb/)
|
||||
|
||||
[Install - Install 3rd party packages](config-install/installpackage_Spark.ipynb/)
|
||||
|
||||
[Restful-Access - Access Spark in BDC via restful Livy APIs](restful-api-accessn/accessing_spark_via_livy.ipynb/)
|
||||
|
||||
## Instructions on how to run in Azure Data Studio
|
||||
|
||||
[data-loading/transform-csv-files.ipynb](dataloading/transform-csv-files.ipynb/)
|
||||
1. From Azure Data Studio Connect to the SQL Server Master instance in a big data cluster.
|
||||
|
||||
2. From Azure Data Studio Connect to the SQL Server Master instance in a big data cluster.
|
||||
2. Right-click on the server name, select **Manage**, switch to **SQL Server Big Data Cluster** tab, and open the notebook in Azure Data Studio. Wait for the “Kernel” and the target context (“Attach to”) to be populated. If required set the relevant “Kernel” ( e.g **PySpark3** ) and **Attach to** needs to be the IP address of your big data cluster endpoint.
|
||||
|
||||
3. Right-click on the server name, select **Manage**, switch to **SQL Server Big Data Cluster** tab, and open the notebook in Azure Data Studio. Wait for the “Kernel” and the target context (“Attach to”) to be populated. If required set the relevant “Kernel” ( e.g **PySpark3** ) and **Attach to** needs to be the IP address of your big data cluster endpoint.
|
||||
|
||||
4. Run each cell in the Notebook sequentially.
|
||||
3. Run each cell in the Notebook sequentially.
|
||||
|
||||
+94
@@ -0,0 +1,94 @@
|
||||
{
|
||||
"metadata": {
|
||||
"kernelspec": {
|
||||
"name": "pyspark3kernel",
|
||||
"display_name": "PySpark3"
|
||||
},
|
||||
"language_info": {
|
||||
"name": "pyspark3",
|
||||
"mimetype": "text/x-python",
|
||||
"codemirror_mode": {
|
||||
"name": "python",
|
||||
"version": 3
|
||||
},
|
||||
"pygments_lexer": "python3"
|
||||
}
|
||||
},
|
||||
"nbformat_minor": 2,
|
||||
"nbformat": 4,
|
||||
"cells": [
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"source": "# Configuring a Spark session using configure-f\r\nRefer to [Spark Configurations](https://spark.apache.org/docs/latest/configuration.html) for specific parameters",
|
||||
"metadata": {}
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"source": "%%configure -f\r\n{\"conf\": {\r\n \"spark.executor.memory\": \"4g\",\r\n \"spark.driver.memory\": \"4g\",\r\n \"spark.executor.cores\": 2,\r\n \"spark.driver.cores\": 1,\r\n \"spark.executor.instances\": 4\r\n }\r\n}",
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"data": {
|
||||
"text/plain": "<IPython.core.display.HTML object>",
|
||||
"text/html": "Current session configs: <tt>{'conf': {'spark.executor.memory': '4g', 'spark.driver.memory': '4g', 'spark.executor.cores': 2, 'spark.driver.cores': 1, 'spark.executor.instances': 4}, 'kind': 'pyspark3'}</tt><br>"
|
||||
},
|
||||
"metadata": {},
|
||||
"output_type": "display_data"
|
||||
},
|
||||
{
|
||||
"data": {
|
||||
"text/plain": "<IPython.core.display.HTML object>",
|
||||
"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>93</td><td>application_1558765999724_0190</td><td>pyspark</td><td>idle</td><td><a target=\"_blank\" href=\"https://10.193.16.144:30443/gateway/default/yarn/proxy/application_1558765999724_0190/\">Link</a></td><td><a target=\"_blank\" href=\"http://storage-0-1.storage-0-svc.test.svc.cluster.local:8042/node/containerlogs/container_1558765999724_0190_01_000001/root\">Link</a></td><td></td></tr></table>"
|
||||
},
|
||||
"metadata": {},
|
||||
"output_type": "display_data"
|
||||
}
|
||||
],
|
||||
"execution_count": 3
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"source": "datafile = \"/spark_data/AdultCensusIncome.csv\"\r\ndf = spark.read.format('csv').options(header='true', inferSchema='true').load(datafile)\r\n\r\ndf.show(5)",
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"name": "stdout",
|
||||
"text": "Starting Spark application\n",
|
||||
"output_type": "stream"
|
||||
},
|
||||
{
|
||||
"data": {
|
||||
"text/plain": "<IPython.core.display.HTML object>",
|
||||
"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>96</td><td>application_1558765999724_0193</td><td>pyspark3</td><td>idle</td><td><a target=\"_blank\" href=\"https://10.193.16.144:30443/gateway/default/yarn/proxy/application_1558765999724_0193/\">Link</a></td><td><a target=\"_blank\" href=\"http://storage-0-0.storage-0-svc.test.svc.cluster.local:8042/node/containerlogs/container_1558765999724_0193_01_000001/root\">Link</a></td><td>✔</td></tr></table>"
|
||||
},
|
||||
"metadata": {},
|
||||
"output_type": "display_data"
|
||||
},
|
||||
{
|
||||
"name": "stdout",
|
||||
"text": "SparkSession available as 'spark'.\n",
|
||||
"output_type": "stream"
|
||||
},
|
||||
{
|
||||
"name": "stdout",
|
||||
"text": "+---+-----------------+--------+----------+--------------+-------------------+------------------+--------------+------+-------+-------------+-------------+---------------+---------------+-------+\n|age| workclass| fnlwgt| education| education-num| marital-status| occupation| relationship| race| sex| capital-gain| capital-loss| hours-per-week| native-country| income|\n+---+-----------------+--------+----------+--------------+-------------------+------------------+--------------+------+-------+-------------+-------------+---------------+---------------+-------+\n| 39| State-gov| 77516.0| Bachelors| 13.0| Never-married| Adm-clerical| Not-in-family| White| Male| 2174.0| 0.0| 40.0| United-States| <=50K|\n| 50| Self-emp-not-inc| 83311.0| Bachelors| 13.0| Married-civ-spouse| Exec-managerial| Husband| White| Male| 0.0| 0.0| 13.0| United-States| <=50K|\n| 38| Private|215646.0| HS-grad| 9.0| Divorced| Handlers-cleaners| Not-in-family| White| Male| 0.0| 0.0| 40.0| United-States| <=50K|\n| 53| Private|234721.0| 11th| 7.0| Married-civ-spouse| Handlers-cleaners| Husband| Black| Male| 0.0| 0.0| 40.0| United-States| <=50K|\n| 28| Private|338409.0| Bachelors| 13.0| Married-civ-spouse| Prof-specialty| Wife| Black| Female| 0.0| 0.0| 40.0| Cuba| <=50K|\n+---+-----------------+--------+----------+--------------+-------------------+------------------+--------------+------+-------+-------------+-------------+---------------+---------------+-------+\nonly showing top 5 rows",
|
||||
"output_type": "stream"
|
||||
}
|
||||
],
|
||||
"execution_count": 4
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"source": "from pyspark import SparkConf\r\nfrom pyspark.sql import SparkSession\r\n\r\ndef isConfiguredItem(cfg_items):\r\n if(cfg_items == 'spark.executor.instances' or cfg_items == 'spark.executor.memory' or \\\r\n cfg_items == 'spark.executor.cores' or cfg_items == 'spark.driver.memory' or \\\r\n cfg_items == 'spark.driver.cores'):\r\n return True\r\n\r\nspark = SparkSession.builder.getOrCreate()\r\nconf = SparkConf().getAll()\r\n\r\nfor cfg_items in conf:\r\n if(isConfiguredItem(cfg_items[0])):\r\n print(cfg_items)\r\n\r\n",
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"name": "stdout",
|
||||
"text": "('spark.executor.instances', '4')\n('spark.driver.memory', '4g')\n('spark.driver.cores', '1')\n('spark.executor.memory', '4g')\n('spark.executor.cores', '2')",
|
||||
"output_type": "stream"
|
||||
}
|
||||
],
|
||||
"execution_count": 21
|
||||
}
|
||||
]
|
||||
}
|
||||
@@ -0,0 +1,91 @@
|
||||
{
|
||||
"metadata": {
|
||||
"kernelspec": {
|
||||
"name": "sparkkernel",
|
||||
"display_name": "Spark | Scala"
|
||||
},
|
||||
"language_info": {
|
||||
"name": "scala",
|
||||
"mimetype": "text/x-scala",
|
||||
"codemirror_mode": "text/x-scala",
|
||||
"pygments_lexer": "scala"
|
||||
}
|
||||
},
|
||||
"nbformat_minor": 2,
|
||||
"nbformat": 4,
|
||||
"cells": [
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"source": "# Packaging in Spark\r\n",
|
||||
"metadata": {}
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"source": "## Use Case 1: I can have key packages in boxed\r\n - All pacakges that come with spark and hadoop distribution\r\n - Python3.5 and Python 2.7\r\n - Pandas, Sklearn and several other supporting ml packages\r\n - R and supporting pacakges as part of MRO\r\n - sparklyr\r\n\r\n \r\n ",
|
||||
"metadata": {}
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"source": "## Use Case 2: I can install pacakges from maven repo to my spark cluster\r\nMaven central is a source of lot of packages. A lot of spark ecosystem pacakges are availble there. These pacakages can be installed to your spark cluster using notebook cell configuration at the start of your spark session.\r\n",
|
||||
"metadata": {}
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"source": "%%configure -f\n{\"conf\": {\"spark.jars.packages\": \"com.microsoft.azure:azure-eventhubs-spark_2.11:2.3.1\"}}",
|
||||
"metadata": {
|
||||
"language": "scala"
|
||||
},
|
||||
"outputs": [
|
||||
{
|
||||
"output_type": "display_data",
|
||||
"data": {
|
||||
"text/plain": "<IPython.core.display.HTML object>",
|
||||
"text/html": "Current session configs: <tt>{'conf': {'spark.jars.packages': 'com.microsoft.azure:azure-eventhubs-spark_2.11:2.3.50'}, 'kind': 'spark'}</tt><br>"
|
||||
},
|
||||
"metadata": {}
|
||||
},
|
||||
{
|
||||
"output_type": "display_data",
|
||||
"data": {
|
||||
"text/plain": "<IPython.core.display.HTML object>",
|
||||
"text/html": "No active sessions."
|
||||
},
|
||||
"metadata": {}
|
||||
}
|
||||
],
|
||||
"execution_count": 3
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"source": "import com.microsoft.azure.eventhubs._",
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"output_type": "stream",
|
||||
"name": "stdout",
|
||||
"text": "import com.microsoft.azure.eventhubs._\n"
|
||||
}
|
||||
],
|
||||
"execution_count": 5
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"source": "## Use Case 3: I have a local jar that i want to run in the spark cluster\r\nAs a user you may build your own customer pacakges that want to run as part of your spark jobs. These pacakges can be uploaded as HDFS and using a notebook configuration spark can consume these pacakges in a jar.\r\n\r\n\r\n",
|
||||
"metadata": {}
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"source": "%%configure -f\r\n {\"conf\": {\"spark.jars\": \"/jar/mycodeJar.jar\"}}",
|
||||
"metadata": {},
|
||||
"outputs": [],
|
||||
"execution_count": 0
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"source": "import com.my.mycodeJar._",
|
||||
"metadata": {},
|
||||
"outputs": [],
|
||||
"execution_count": 0
|
||||
}
|
||||
]
|
||||
}
|
||||
BIN
Binary file not shown.
|
After Width: | Height: | Size: 221 KiB |
+150
@@ -0,0 +1,150 @@
|
||||
{
|
||||
"metadata": {
|
||||
"kernelspec": {
|
||||
"name": "pyspark3kernel",
|
||||
"display_name": "PySpark3"
|
||||
},
|
||||
"language_info": {
|
||||
"name": "pyspark3",
|
||||
"mimetype": "text/x-python",
|
||||
"codemirror_mode": {
|
||||
"name": "python",
|
||||
"version": 3
|
||||
},
|
||||
"pygments_lexer": "python3"
|
||||
}
|
||||
},
|
||||
"nbformat_minor": 2,
|
||||
"nbformat": 4,
|
||||
"cells": [
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"source": "# Read and write from Spark to SQL using the MSSQL Spark Connector\r\nA typical big data scenario a key usage pattern is high volume, velocity and variety data processing in Spark followed with batch or streaming writes to SQL for access to LOB applications. These usage patterns greatly benefit from a connector that utilizes key SQL optimizations and provides an efficient write to SQLServer master instance and SQL Server data pool in Big Data Clusters.\r\n\r\nMSSQL Spark connector provides an efficient write SQLServer master instance and SQL Server data pool in Big Data Clusters.\r\n\r\nUsage\r\n----\r\n- Familiar Spark DataSource V1 interface\r\n- Referenced by the name \"mssql\" or fully qualified \"com.microsoft.sqlserver.jdbc.spark\"\r\n- Use from supported Spark language bindings ( Python, Scala, Java, R)\r\n- Optionally pass Bulk Copy parameters \r\n\r\n** Note : The image here may not be visible dues to markdown bug. Please change path here to full path to view the image.\r\n<img src =\r\n\"../data-virtualization/MSSQL_Spark_Connector2.jpg\" style=\"float: center;\" alt=\"drawing\" width=\"900\">\r\n\r\nMore details\r\n-----------\r\n\r\nMSSQL Spark connector, uses [SQL Server Bulk copy APIS](https://docs.microsoft.com/en-us/sql/connect/jdbc/using-bulk-copy-with-the-jdbc-driver?view=sql-server-2017#sqlserverbulkcopyoptions) to implement an efficient write to SQL Server. The connector is based on Spark Data source APIs and provides a familiar JDBC interface for access\r\n\r\nThe Sample\r\n---------\r\n\r\nThe following sample shows MSSQL JDBC Connector for read/write QLServer master instance and SQL Server data pool in Big Data Clusters. The sample is divided into 2 parts. The first part shows read/write to SQL Master instance and Part 2 shows read/write to Data Pools in Big Data Cluster. \r\n\r\nIn the sample we' ll \r\n- Read a file from HDFS and do some basic processing \r\n- In Part 1, we'll write the dataframe to SQL server table and then read the table to a dataframe .\r\n- In Part 2, we'll write the dataframe to SQL Server data pool external table and then read it back to a spark data frame. \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 spark_data and upload the file there. \r\n- [For CTP2.5] Configure the spark session to use the MSSQL Connector jar. The jar can be found at /jar/spark-mssql-connector-assembly-1.0.0.jar post deployment of Big Data Cluster.\r\n\r\n``` sh\r\n %%configure -f\r\n {\"conf\": {\"spark.jars\": \"/jar/spark-mssql-connector-assembly-1.0.0.jar\"}}\r\n```\r\n\r\n\r\n ",
|
||||
"metadata": {}
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"source": "# Read CSV into a data frame\r\nIn this step we read the CSV into a data frame and do some basic cleanup steps. \r\n\r\n\r\n",
|
||||
"metadata": {}
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"source": "#spark = SparkSession.builder.getOrCreate()\r\nsc.setLogLevel(\"INFO\")\r\n\r\n#Read a file and then write it to the SQL table\r\ndatafile = \"/spark_data/AdultCensusIncome.csv\"\r\ndf = spark.read.format('csv').options(header='true', inferSchema='true', ignoreLeadingWhiteSpace='true', ignoreTrailingWhiteSpace='true').load(datafile)\r\ndf.show(5)\r\n\r\n\r\n#Process this data. Very simple data cleanup steps. Replacing \"-\" with \"_\" in column names\r\ncolumns_new = [col.replace(\"-\", \"_\") for col in df.columns]\r\ndf = df.toDF(*columns_new)\r\ndf.show(5)\r\n",
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"output_type": "stream",
|
||||
"name": "stdout",
|
||||
"text": "Starting Spark application\n"
|
||||
},
|
||||
{
|
||||
"output_type": "display_data",
|
||||
"data": {
|
||||
"text/plain": "<IPython.core.display.HTML object>",
|
||||
"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>0</td><td>application_1561072870175_0001</td><td>pyspark3</td><td>idle</td><td><a target=\"_blank\" href=\"https://40.78.87.105:30443/gateway/default/yarn/proxy/application_1561072870175_0001/\">Link</a></td><td><a target=\"_blank\" href=\"http://storage-0-0.storage-0-svc.testbdc.svc.cluster.local:8042/node/containerlogs/container_1561072870175_0001_01_000001/root\">Link</a></td><td>✔</td></tr></table>"
|
||||
},
|
||||
"metadata": {}
|
||||
},
|
||||
{
|
||||
"output_type": "stream",
|
||||
"name": "stdout",
|
||||
"text": "SparkSession available as 'spark'.\n"
|
||||
},
|
||||
{
|
||||
"output_type": "stream",
|
||||
"name": "stdout",
|
||||
"text": "+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\n|age| workclass|fnlwgt|education|education-num| marital-status| occupation| relationship| race| sex|capital-gain|capital-loss|hours-per-week|native-country|income|\n+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\n| 39| State-gov| 77516|Bachelors| 13| Never-married| Adm-clerical|Not-in-family|White| Male| 2174| 0| 40| United-States| <=50K|\n| 50|Self-emp-not-inc| 83311|Bachelors| 13|Married-civ-spouse| Exec-managerial| Husband|White| Male| 0| 0| 13| United-States| <=50K|\n| 38| Private|215646| HS-grad| 9| Divorced|Handlers-cleaners|Not-in-family|White| Male| 0| 0| 40| United-States| <=50K|\n| 53| Private|234721| 11th| 7|Married-civ-spouse|Handlers-cleaners| Husband|Black| Male| 0| 0| 40| United-States| <=50K|\n| 28| Private|338409|Bachelors| 13|Married-civ-spouse| Prof-specialty| Wife|Black|Female| 0| 0| 40| Cuba| <=50K|\n+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\nonly showing top 5 rows\n\n+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\n|age| workclass|fnlwgt|education|education_num| marital_status| occupation| relationship| race| sex|capital_gain|capital_loss|hours_per_week|native_country|income|\n+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\n| 39| State-gov| 77516|Bachelors| 13| Never-married| Adm-clerical|Not-in-family|White| Male| 2174| 0| 40| United-States| <=50K|\n| 50|Self-emp-not-inc| 83311|Bachelors| 13|Married-civ-spouse| Exec-managerial| Husband|White| Male| 0| 0| 13| United-States| <=50K|\n| 38| Private|215646| HS-grad| 9| Divorced|Handlers-cleaners|Not-in-family|White| Male| 0| 0| 40| United-States| <=50K|\n| 53| Private|234721| 11th| 7|Married-civ-spouse|Handlers-cleaners| Husband|Black| Male| 0| 0| 40| United-States| <=50K|\n| 28| Private|338409|Bachelors| 13|Married-civ-spouse| Prof-specialty| Wife|Black|Female| 0| 0| 40| Cuba| <=50K|\n+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\nonly showing top 5 rows"
|
||||
}
|
||||
],
|
||||
"execution_count": 3
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"source": "# (PART 1) Write and READ to SQL Table\r\n- Write dataframe to SQL table to SQL Master\r\n- Read SQL Table to Spark dataframe",
|
||||
"metadata": {}
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"source": "#Write from Spark to SQL table using MSSQL Spark Connector\r\nprint(\"Use MSSQL connector to write to master SQL instance \")\r\n\r\nservername = \"jdbc:sqlserver://master-0.master-svc\"\r\ndbname = \"MyTestDatabase\"\r\nurl = servername + \";\" + \"databaseName=\" + dbname + \";\"\r\n\r\ndbtable = \"AdultCensus_test\"\r\nuser = \"*****\"\r\npassword = \"*****\" # Please specify password here\r\n\r\n#com.microsoft.sqlserver.jdbc.spark\r\n\r\ntry:\r\n df.write \\\r\n .format(\"com.microsoft.sqlserver.jdbc.spark\") \\\r\n .mode(\"overwrite\") \\\r\n .option(\"url\", url) \\\r\n .option(\"dbtable\", dbtable) \\\r\n .option(\"user\", user) \\\r\n .option(\"password\", password) \\\r\n .save()\r\nexcept ValueError as error :\r\n print(\"MSSQL Connector write failed\", error)\r\n\r\nprint(\"MSSQL Connector write(overwrite) succeeded \")\r\n\r\n\r\n",
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"output_type": "stream",
|
||||
"name": "stdout",
|
||||
"text": "Use MSSQL connector to write to master SQL instance \nMSSQL Connector write(overwrite) succeeded"
|
||||
}
|
||||
],
|
||||
"execution_count": 5
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"source": "#Use mode as append\r\ntry:\r\n df.write \\\r\n .format(\"com.microsoft.sqlserver.jdbc.spark\") \\\r\n .mode(\"append\") \\\r\n .option(\"url\", url) \\\r\n .option(\"dbtable\", dbtable) \\\r\n .option(\"user\", user) \\\r\n .option(\"password\", password) \\\r\n .save()\r\nexcept ValueError as error :\r\n print(\"MSSQL Connector write failed\", error)\r\n\r\nprint(\"MSSQL Connector write(append) succeeded \")",
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"output_type": "stream",
|
||||
"name": "stdout",
|
||||
"text": "MSSQL Connector write(append) succeeded"
|
||||
}
|
||||
],
|
||||
"execution_count": 6
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"source": "#Read from SQL table using MSSQ Connector\r\nprint(\"read data from SQL server table \")\r\njdbcDF = 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).load()\r\n\r\njdbcDF.show(5)",
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"output_type": "stream",
|
||||
"name": "stdout",
|
||||
"text": "read data from SQL server table \n+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\n|age| workclass|fnlwgt|education|education_num| marital_status| occupation| relationship| race| sex|capital_gain|capital_loss|hours_per_week|native_country|income|\n+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\n| 39| State-gov| 77516|Bachelors| 13| Never-married| Adm-clerical|Not-in-family|White| Male| 2174| 0| 40| United-States| <=50K|\n| 50|Self-emp-not-inc| 83311|Bachelors| 13|Married-civ-spouse| Exec-managerial| Husband|White| Male| 0| 0| 13| United-States| <=50K|\n| 38| Private|215646| HS-grad| 9| Divorced|Handlers-cleaners|Not-in-family|White| Male| 0| 0| 40| United-States| <=50K|\n| 53| Private|234721| 11th| 7|Married-civ-spouse|Handlers-cleaners| Husband|Black| Male| 0| 0| 40| United-States| <=50K|\n| 28| Private|338409|Bachelors| 13|Married-civ-spouse| Prof-specialty| Wife|Black|Female| 0| 0| 40| Cuba| <=50K|\n+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\nonly showing top 5 rows"
|
||||
}
|
||||
],
|
||||
"execution_count": 7
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"source": "# (PART 2) Write and READ to Data Pool external Tables in Big Data Cluster\r\n- Write dataframe to SQL external table in Data Pools in Big Data Cluste\r\n- Read SQL external Table to Spark dataframe",
|
||||
"metadata": {}
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"source": "#Write from Spark to SQL table using MSSQL Spark Connector\r\nprint(\"Use MSSQL connector to write to master SQL instance \")\r\n\r\ndatapool_table = \"AdultCensus_DataPoolTable\"\r\ndatasource_name = \"test_data_src\"\r\n\r\n\r\ntry:\r\n df.write \\\r\n .format(\"com.microsoft.sqlserver.jdbc.spark\") \\\r\n .mode(\"overwrite\") \\\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)\\\r\n .save()\r\nexcept ValueError as error :\r\n print(\"MSSQL Connector write failed\", error)\r\n\r\nprint(\"MSSQL Connector write(overwrite) to data pool external table succeeded\")\r\n",
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"output_type": "stream",
|
||||
"name": "stdout",
|
||||
"text": "Use MSSQL connector to write to master SQL instance \nMSSQL Connector write(overwrite) to data pool external table succeeded"
|
||||
}
|
||||
],
|
||||
"execution_count": 8
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"source": "try:\r\n df.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)\\\r\n .save()\r\nexcept ValueError as error :\r\n print(\"MSSQL Connector write failed\", error)\r\n\r\nprint(\"MSSQL Connector write(append) to data pool external table succeeded\")",
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"output_type": "stream",
|
||||
"name": "stdout",
|
||||
"text": "MSSQL Connector write(append) to data pool external table succeeded"
|
||||
}
|
||||
],
|
||||
"execution_count": 9
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"source": "#Read from SQL table using MSSQ Connector\r\nprint(\"read data from SQL server table \")\r\njdbcDF = spark.read \\\r\n .format(\"com.microsoft.sqlserver.jdbc.spark\") \\\r\n .option(\"url\", url) \\\r\n .option(\"dbtable\", datapool_table) \\\r\n .option(\"user\", user) \\\r\n .option(\"password\", password)\\\r\n .load()\r\n\r\njdbcDF.show(5)\r\n\r\nprint(\"MSSQL Connector read from data pool external table succeeded\")",
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"output_type": "stream",
|
||||
"name": "stdout",
|
||||
"text": "read data from SQL server table \n+---+----------------+------+----------+-------------+------------------+---------------+-------------+-----+------+------------+------------+--------------+--------------+------+\n|age| workclass|fnlwgt| education|education_num| marital_status| occupation| relationship| race| sex|capital_gain|capital_loss|hours_per_week|native_country|income|\n+---+----------------+------+----------+-------------+------------------+---------------+-------------+-----+------+------------+------------+--------------+--------------+------+\n| 46|Self-emp-not-inc|277946|Assoc-acdm| 12| Separated| Craft-repair|Not-in-family|White| Male| 0| 0| 40| United-States| <=50K|\n| 38| Private| 91039| Bachelors| 13|Married-civ-spouse| Sales| Husband|White| Male| 15024| 0| 60| United-States| >50K|\n| 18| Private|156764| 11th| 7| Never-married| Other-service| Own-child|White| Male| 0| 0| 40| United-States| <=50K|\n| 34| Private|136721| HS-grad| 9| Divorced|Exec-managerial|Not-in-family|White|Female| 0| 0| 40| United-States| <=50K|\n| 39| State-gov| 77516| Bachelors| 13| Never-married| Adm-clerical|Not-in-family|White| Male| 2174| 0| 40| United-States| <=50K|\n+---+----------------+------+----------+-------------+------------------+---------------+-------------+-----+------+------------+------------+--------------+--------------+------+\nonly showing top 5 rows\n\nMSSQL Connector read from data pool external table succeeded"
|
||||
}
|
||||
],
|
||||
"execution_count": 10
|
||||
}
|
||||
]
|
||||
}
|
||||
+15
-15
@@ -28,27 +28,27 @@
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"output_type": "stream",
|
||||
"name": "stdout",
|
||||
"text": "Starting Spark application\n",
|
||||
"output_type": "stream"
|
||||
"text": "Starting Spark application\n"
|
||||
},
|
||||
{
|
||||
"output_type": "display_data",
|
||||
"data": {
|
||||
"text/plain": "<IPython.core.display.HTML object>",
|
||||
"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>2</td><td>application_1554755839506_0003</td><td>pyspark3</td><td>idle</td><td><a target=\"_blank\" href=\"https://40.78.42.207:30443/gateway/default/yarn/proxy/application_1554755839506_0003/\">Link</a></td><td><a target=\"_blank\" href=\"http://storage-0-1.storage-0-svc.mssql-cluster.svc.cluster.local:8042/node/containerlogs/container_1554755839506_0003_01_000001/root\">Link</a></td><td>✔</td></tr></table>"
|
||||
},
|
||||
"metadata": {},
|
||||
"output_type": "display_data"
|
||||
"metadata": {}
|
||||
},
|
||||
{
|
||||
"output_type": "stream",
|
||||
"name": "stdout",
|
||||
"text": "SparkSession available as 'spark'.\n",
|
||||
"output_type": "stream"
|
||||
"text": "SparkSession available as 'spark'.\n"
|
||||
},
|
||||
{
|
||||
"output_type": "stream",
|
||||
"name": "stdout",
|
||||
"text": "+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\n|age| workclass|fnlwgt|education|education-num| marital-status| occupation| relationship| race| sex|capital-gain|capital-loss|hours-per-week|native-country|income|\n+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\n| 39| State-gov| 77516|Bachelors| 13| Never-married| Adm-clerical|Not-in-family|White| Male| 2174| 0| 40| United-States| <=50K|\n| 50|Self-emp-not-inc| 83311|Bachelors| 13|Married-civ-spouse| Exec-managerial| Husband|White| Male| 0| 0| 13| United-States| <=50K|\n| 38| Private|215646| HS-grad| 9| Divorced|Handlers-cleaners|Not-in-family|White| Male| 0| 0| 40| United-States| <=50K|\n| 53| Private|234721| 11th| 7|Married-civ-spouse|Handlers-cleaners| Husband|Black| Male| 0| 0| 40| United-States| <=50K|\n| 28| Private|338409|Bachelors| 13|Married-civ-spouse| Prof-specialty| Wife|Black|Female| 0| 0| 40| Cuba| <=50K|\n+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\nonly showing top 5 rows",
|
||||
"output_type": "stream"
|
||||
"text": "+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\n|age| workclass|fnlwgt|education|education-num| marital-status| occupation| relationship| race| sex|capital-gain|capital-loss|hours-per-week|native-country|income|\n+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\n| 39| State-gov| 77516|Bachelors| 13| Never-married| Adm-clerical|Not-in-family|White| Male| 2174| 0| 40| United-States| <=50K|\n| 50|Self-emp-not-inc| 83311|Bachelors| 13|Married-civ-spouse| Exec-managerial| Husband|White| Male| 0| 0| 13| United-States| <=50K|\n| 38| Private|215646| HS-grad| 9| Divorced|Handlers-cleaners|Not-in-family|White| Male| 0| 0| 40| United-States| <=50K|\n| 53| Private|234721| 11th| 7|Married-civ-spouse|Handlers-cleaners| Husband|Black| Male| 0| 0| 40| United-States| <=50K|\n| 28| Private|338409|Bachelors| 13|Married-civ-spouse| Prof-specialty| Wife|Black|Female| 0| 0| 40| Cuba| <=50K|\n+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\nonly showing top 5 rows"
|
||||
}
|
||||
],
|
||||
"execution_count": 3
|
||||
@@ -59,22 +59,22 @@
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"output_type": "stream",
|
||||
"name": "stdout",
|
||||
"text": "+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\n|age| workclass|fnlwgt|education|education_num| marital_status| occupation| relationship| race| sex|capital_gain|capital_loss|hours_per_week|native_country|income|\n+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\n| 39| State-gov| 77516|Bachelors| 13| Never-married| Adm-clerical|Not-in-family|White| Male| 2174| 0| 40| United-States| <=50K|\n| 50|Self-emp-not-inc| 83311|Bachelors| 13|Married-civ-spouse| Exec-managerial| Husband|White| Male| 0| 0| 13| United-States| <=50K|\n| 38| Private|215646| HS-grad| 9| Divorced|Handlers-cleaners|Not-in-family|White| Male| 0| 0| 40| United-States| <=50K|\n| 53| Private|234721| 11th| 7|Married-civ-spouse|Handlers-cleaners| Husband|Black| Male| 0| 0| 40| United-States| <=50K|\n| 28| Private|338409|Bachelors| 13|Married-civ-spouse| Prof-specialty| Wife|Black|Female| 0| 0| 40| Cuba| <=50K|\n+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\nonly showing top 5 rows",
|
||||
"output_type": "stream"
|
||||
"text": "+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\n|age| workclass|fnlwgt|education|education_num| marital_status| occupation| relationship| race| sex|capital_gain|capital_loss|hours_per_week|native_country|income|\n+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\n| 39| State-gov| 77516|Bachelors| 13| Never-married| Adm-clerical|Not-in-family|White| Male| 2174| 0| 40| United-States| <=50K|\n| 50|Self-emp-not-inc| 83311|Bachelors| 13|Married-civ-spouse| Exec-managerial| Husband|White| Male| 0| 0| 13| United-States| <=50K|\n| 38| Private|215646| HS-grad| 9| Divorced|Handlers-cleaners|Not-in-family|White| Male| 0| 0| 40| United-States| <=50K|\n| 53| Private|234721| 11th| 7|Married-civ-spouse|Handlers-cleaners| Husband|Black| Male| 0| 0| 40| United-States| <=50K|\n| 28| Private|338409|Bachelors| 13|Married-civ-spouse| Prof-specialty| Wife|Black|Female| 0| 0| 40| Cuba| <=50K|\n+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\nonly showing top 5 rows"
|
||||
}
|
||||
],
|
||||
"execution_count": 4
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"source": "#Write from Spark to SQL table using JDBC\r\nprint(\"Use build in JDBC connector to write to SQLServer master instance in Big data \")\r\n\r\nservername = \"jdbc:sqlserver://master-0.master-svc\"\r\ndbname = \"MyTestDatabase\"\r\nurl = servername + \";\" + \"databaseName=\" + dbname + \";\"\r\n\r\ndbtable = \"dbo.AdultCensus\"\r\nuser = \"sa\"\r\npassword = \"Yukon900\"\r\n\r\nprint(\"url is \", url)\r\n\r\ntry:\r\n df.write \\\r\n .format(\"jdbc\") \\\r\n .mode(\"overwrite\") \\\r\n .option(\"url\", url) \\\r\n .option(\"dbtable\", dbtable) \\\r\n .option(\"user\", user) \\\r\n .option(\"password\", password)\\\r\n .save()\r\nexcept ValueError as error :\r\n print(\"JDBC Write failed\", error)\r\n\r\nprint(\"JDBC Write done \")\r\n\r\n\r\n",
|
||||
"source": "#Write from Spark to SQL table using JDBC\r\nprint(\"Use build in JDBC connector to write to SQLServer master instance in Big data \")\r\n\r\nservername = \"jdbc:sqlserver://master-0.master-svc\"\r\ndbname = \"MyTestDatabase\"\r\nurl = servername + \";\" + \"databaseName=\" + dbname + \";\"\r\n\r\ndbtable = \"dbo.AdultCensus\"\r\nuser = \"***\"\r\npassword = \"****\"\r\n\r\nprint(\"url is \", url)\r\n\r\ntry:\r\n df.write \\\r\n .format(\"jdbc\") \\\r\n .mode(\"overwrite\") \\\r\n .option(\"url\", url) \\\r\n .option(\"dbtable\", dbtable) \\\r\n .option(\"user\", user) \\\r\n .option(\"password\", password)\\\r\n .save()\r\nexcept ValueError as error :\r\n print(\"JDBC Write failed\", error)\r\n\r\nprint(\"JDBC Write done \")\r\n\r\n\r\n",
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"output_type": "stream",
|
||||
"name": "stdout",
|
||||
"text": "Use build in JDBC connector to write to SQLServer master instance in Big data \nurl is jdbc:sqlserver://master-0.master-svc;databaseName=MyTestDatabase;\nJDBC Write done",
|
||||
"output_type": "stream"
|
||||
"text": "Use build in JDBC connector to write to SQLServer master instance in Big data \nurl is jdbc:sqlserver://master-0.master-svc;databaseName=MyTestDatabase;\nJDBC Write done"
|
||||
}
|
||||
],
|
||||
"execution_count": 9
|
||||
@@ -85,9 +85,9 @@
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"output_type": "stream",
|
||||
"name": "stdout",
|
||||
"text": "read data from SQL server table \n+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\n|age| workclass|fnlwgt|education|education_num| marital_status| occupation| relationship| race| sex|capital_gain|capital_loss|hours_per_week|native_country|income|\n+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\n| 39| State-gov| 77516|Bachelors| 13| Never-married| Adm-clerical|Not-in-family|White| Male| 2174| 0| 40| United-States| <=50K|\n| 50|Self-emp-not-inc| 83311|Bachelors| 13|Married-civ-spouse| Exec-managerial| Husband|White| Male| 0| 0| 13| United-States| <=50K|\n| 38| Private|215646| HS-grad| 9| Divorced|Handlers-cleaners|Not-in-family|White| Male| 0| 0| 40| United-States| <=50K|\n| 53| Private|234721| 11th| 7|Married-civ-spouse|Handlers-cleaners| Husband|Black| Male| 0| 0| 40| United-States| <=50K|\n| 28| Private|338409|Bachelors| 13|Married-civ-spouse| Prof-specialty| Wife|Black|Female| 0| 0| 40| Cuba| <=50K|\n+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\nonly showing top 5 rows",
|
||||
"output_type": "stream"
|
||||
"text": "read data from SQL server table \n+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\n|age| workclass|fnlwgt|education|education_num| marital_status| occupation| relationship| race| sex|capital_gain|capital_loss|hours_per_week|native_country|income|\n+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\n| 39| State-gov| 77516|Bachelors| 13| Never-married| Adm-clerical|Not-in-family|White| Male| 2174| 0| 40| United-States| <=50K|\n| 50|Self-emp-not-inc| 83311|Bachelors| 13|Married-civ-spouse| Exec-managerial| Husband|White| Male| 0| 0| 13| United-States| <=50K|\n| 38| Private|215646| HS-grad| 9| Divorced|Handlers-cleaners|Not-in-family|White| Male| 0| 0| 40| United-States| <=50K|\n| 53| Private|234721| 11th| 7|Married-civ-spouse|Handlers-cleaners| Husband|Black| Male| 0| 0| 40| United-States| <=50K|\n| 28| Private|338409|Bachelors| 13|Married-civ-spouse| Prof-specialty| Wife|Black|Female| 0| 0| 40| Cuba| <=50K|\n+---+----------------+------+---------+-------------+------------------+-----------------+-------------+-----+------+------------+------------+--------------+--------------+------+\nonly showing top 5 rows"
|
||||
}
|
||||
],
|
||||
"execution_count": 11
|
||||
|
||||
+188
@@ -0,0 +1,188 @@
|
||||
{
|
||||
"metadata": {
|
||||
"kernelspec": {
|
||||
"name": "python3",
|
||||
"display_name": "Python 3"
|
||||
},
|
||||
"language_info": {
|
||||
"name": "python",
|
||||
"version": "3.6.6",
|
||||
"mimetype": "text/x-python",
|
||||
"codemirror_mode": {
|
||||
"name": "ipython",
|
||||
"version": 3
|
||||
},
|
||||
"pygments_lexer": "ipython3",
|
||||
"nbconvert_exporter": "python",
|
||||
"file_extension": ".py"
|
||||
}
|
||||
},
|
||||
"nbformat_minor": 2,
|
||||
"nbformat": 4,
|
||||
"cells": [
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"source": "# Using Spark from Livy end point\r\nThe samples demostrates using spark service via the livy end point. The sample uses the python request library to access the livy interface. \r\nLivy API details are available at https://livy.incubator.apache.org/docs/latest/rest-api.html\r\n** Note : The image here may not be visible dues to markdown bug. Please change path here to full path to view the image.\r\n<img src = \"./spark_from_livy.jpg\" style=\"float: center;\" alt=\"drawing\" width=\"900\">\r\n\r\n",
|
||||
"metadata": {}
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"source": "## Using python requests library to access restful APIs",
|
||||
"metadata": {}
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"source": "import json, pprint, requests, textwrap\r\nfrom requests.auth import HTTPBasicAuth\r\nprint(\"Get username and email \")\r\nprint(\"---------------------- \")\r\n\r\n#Basic test for request\r\nimport requests\r\n\r\nr = requests.get('https://jsonplaceholder.typicode.com/users')\r\nr.encoding = 'utf-8'\r\nret_json = r.json()\r\n\r\n\r\nfor user in ret_json:\r\n print(user['username'] , user['email'])\r\n\r\n#print(r.json()) \r\n\r\n",
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"output_type": "stream",
|
||||
"name": "stdout",
|
||||
"text": "Get username and email \n---------------------- \n"
|
||||
},
|
||||
{
|
||||
"output_type": "stream",
|
||||
"name": "stdout",
|
||||
"text": "Bret Sincere@april.biz\nAntonette Shanna@melissa.tv\nSamantha Nathan@yesenia.net\nKarianne Julianne.OConner@kory.org\nKamren Lucio_Hettinger@annie.ca\nLeopoldo_Corkery Karley_Dach@jasper.info\nElwyn.Skiles Telly.Hoeger@billy.biz\nMaxime_Nienow Sherwood@rosamond.me\nDelphine Chaim_McDermott@dana.io\nMoriah.Stanton Rey.Padberg@karina.biz\n"
|
||||
}
|
||||
],
|
||||
"execution_count": 1
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"source": "## Livy APIs to fetch running sessions and sesssion state",
|
||||
"metadata": {}
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"source": "import json, pprint, requests, textwrap\r\nfrom requests.auth import HTTPBasicAuth\r\n\r\n#Diabling exception to avoid the https verificatin warning\r\nimport requests.packages.urllib3 as urllib3\r\nimport urllib3.exceptions as urllib3_exceptions \r\nurllib3.disable_warnings(urllib3_exceptions.InsecureRequestWarning)\r\n\r\n\r\n#Change host per your confirgration \r\nhost = \"https://<ip>:<port>/gateway/default/livy/v1\"\r\n\r\n#Livy interface as per https://livy.incubator.apache.org/docs/latest/rest-api.html#session\r\n\r\n# Construct Request - Get a list of current spark sessions.\r\nsessions_url = host + \"/sessions\"\r\n\r\n# Common headers for all requests.\r\n# Auth header\r\nauth = HTTPBasicAuth(\"***\", \"***\")\r\n# Content Type\r\nheaders = {'Content-Type': 'application/json'}\r\n\r\ndata = {\r\n 'from': 0, \r\n 'size': 10\r\n }\r\nr = requests.get(sessions_url, data=json.dumps(data), headers=headers, auth=auth, verify=False)\r\n\r\nresponse_body = r.json()\r\n\r\nprint(\"Sessions fetched starting \", response_body['from'])\r\nprint(\"Number of session fetched\", response_body['total'])\r\nsession_list = response_body['sessions']\r\n\r\nfor session in session_list:\r\n print(\"The session {0} has a state {1} \".format(session['id'],session['state']))\r\n\r\n",
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"output_type": "stream",
|
||||
"name": "stdout",
|
||||
"text": "Sessions fetched starting 0\nNumber of session fetched 1\nThe session 10 has a state idle \n"
|
||||
}
|
||||
],
|
||||
"execution_count": 2
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"source": "## Create a new Spark session and query state",
|
||||
"metadata": {}
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"source": "#Create a new Spark session\r\ndata = {\r\n 'kind':'pyspark'\r\n}\r\n\r\nr = requests.post(sessions_url, data=json.dumps(data), headers=headers, auth=auth, verify=False)\r\nresponse_body = r.json()\r\n\r\nsession_id = response_body['id']\r\nsession_state = response_body['state']\r\nprint(\"Spark session {0} created. Current state is {1}\".format(session_id,session_state))\r\ncreated_session_url = r.headers['location']\r\n\r\n\r\n\r\n",
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"output_type": "stream",
|
||||
"name": "stdout",
|
||||
"text": "Spark session 13 created. Current state is starting\n"
|
||||
}
|
||||
],
|
||||
"execution_count": 3
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"source": "#Query information about the session we just created\r\nthis_session_url = host + created_session_url\r\nprint(\"this_session_url\", this_session_url)\r\n\r\nr = requests.get(this_session_url, headers=headers,auth=auth, verify=False)\r\npprint.pprint(r.json())",
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"output_type": "stream",
|
||||
"name": "stdout",
|
||||
"text": "this_session_url https://13.91.32.53:30443/gateway/default/livy/v1/sessions/13\n"
|
||||
},
|
||||
{
|
||||
"output_type": "stream",
|
||||
"name": "stdout",
|
||||
"text": "{'appId': 'application_1560270944894_0014',\n 'appInfo': {'driverLogUrl': 'http://storage-0-0.storage-0-svc.newaks.svc.cluster.local:8042/node/containerlogs/container_1560270944894_0014_01_000001/root',\n 'sparkUiUrl': 'http://master-0.master-svc:8088/proxy/application_1560270944894_0014/'},\n 'id': 13,\n 'kind': 'pyspark',\n 'log': ['\\t ApplicationMaster RPC port: -1',\n '\\t queue: default',\n '\\t start time: 1560880472284',\n '\\t final status: UNDEFINED',\n '\\t tracking URL: '\n 'http://master-0.master-svc:8088/proxy/application_1560270944894_0014/',\n '\\t user: root',\n '19/06/18 17:54:32 INFO util.ShutdownHookManager: Shutdown hook '\n 'called',\n '19/06/18 17:54:32 INFO util.ShutdownHookManager: Deleting directory '\n '/tmp/spark-de7b25a4-1605-4d7f-bef5-c921175c8ae9',\n '19/06/18 17:54:32 INFO util.ShutdownHookManager: Deleting directory '\n '/tmp/spark-00164108-299d-4313-b800-04bbfb8c1e5a',\n '\\nYARN Diagnostics: '],\n 'name': None,\n 'owner': None,\n 'proxyUser': None,\n 'state': 'idle'}\n"
|
||||
}
|
||||
],
|
||||
"execution_count": 11
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"source": "## Execute code interactively Spark session and check results",
|
||||
"metadata": {}
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"source": "\r\n#Execute code interactively in this session using session/<ID>/statements interface\r\n\r\nstatements_url = this_session_url + \"/statements\"\r\npprint.pprint(statements_url)\r\n\r\ndata = {\r\n 'code': \"11+11\"\r\n}\r\n\r\nr = requests.post(statements_url, data=json.dumps(data), headers=headers, auth=auth, verify=False)\r\nprint(\"Respone status code\" ,r.status_code )\r\nprint(\"Response received is \",r.json())\r\nprint(\"Poll URI is \",r.headers['location'])\r\n#response_body = r.json()\r\n\r\n\r\n\r\n",
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"output_type": "stream",
|
||||
"name": "stdout",
|
||||
"text": "'https://13.91.32.53:30443/gateway/default/livy/v1/sessions/13/statements'\n"
|
||||
},
|
||||
{
|
||||
"output_type": "stream",
|
||||
"name": "stdout",
|
||||
"text": "Respone status code 201\nResponse received is {'id': 0, 'code': '11+11', 'state': 'waiting', 'output': None, 'progress': 0.0}\nPoll URI is /sessions/13/statements/0\n"
|
||||
}
|
||||
],
|
||||
"execution_count": 12
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"source": "# Wait a while before executing this cell. Spark session start up takes time to run your code.\r\nspecific_statement = host + r.headers['location']\r\nprint(\"monitoring url is \", specific_statement)\r\n\r\nr = requests.get(specific_statement, headers=headers, auth=auth, verify=False)\r\nprint(\"Response status is \",r.status_code)\r\nprint(\"Response received is \", pprint.pprint(r.json()))",
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"output_type": "stream",
|
||||
"name": "stdout",
|
||||
"text": "monitoring url is https://13.91.32.53:30443/gateway/default/livy/v1/sessions/13/statements/0\n"
|
||||
},
|
||||
{
|
||||
"output_type": "stream",
|
||||
"name": "stdout",
|
||||
"text": "Response status is 200\n{'code': '11+11',\n 'id': 0,\n 'output': {'data': {'text/plain': '22'}, 'execution_count': 0, 'status': 'ok'},\n 'progress': 1.0,\n 'state': 'available'}\nResponse received is None\n"
|
||||
}
|
||||
],
|
||||
"execution_count": 13
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"source": "## Execute code in batch",
|
||||
"metadata": {}
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"source": "#Execute code in batch\r\n#The following uses pi.py from spark source. Please get that file and transer to HDFS /jar folder.\r\nbatch_url = host + \"/batches\"\r\nprint(\"batch_url\", batch_url)\r\n\r\ndata = {\r\n 'file' : '/jar/pi.py' \r\n}\r\n\r\nr = requests.post(batch_url, data=json.dumps(data), headers=headers, auth=auth, verify=False)\r\nreturned_batch_url = r.headers['location']\r\nprint(\"Respone status code\" , r.status_code)\r\nprint(\"Poll URI is \",returned_batch_url )\r\nprint(\"Response is \", pprint.pprint(r.json()))",
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"output_type": "stream",
|
||||
"name": "stdout",
|
||||
"text": "batch_url https://13.91.32.53:30443/gateway/default/livy/v1/batches\n"
|
||||
},
|
||||
{
|
||||
"output_type": "stream",
|
||||
"name": "stdout",
|
||||
"text": "Respone status code 201\nPoll URI is /batches/1\n{'appId': None,\n 'appInfo': {'driverLogUrl': None, 'sparkUiUrl': None},\n 'id': 1,\n 'log': ['stdout: ', '\\nstderr: ', '\\nYARN Diagnostics: '],\n 'name': None,\n 'state': 'starting'}\nResponse is None\n"
|
||||
}
|
||||
],
|
||||
"execution_count": 15
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"source": "#Check results of executed code\r\nspecific_batch = host + returned_batch_url\r\nprint(\"specific batch request \",specific_batch)\r\n\r\nr = requests.get(specific_batch,headers=headers, auth=auth, verify = False)\r\nprint(\"Response status is \",r.status_code)\r\nprint(\"Response received is \", pprint.pprint(r.json()))",
|
||||
"metadata": {},
|
||||
"outputs": [
|
||||
{
|
||||
"output_type": "stream",
|
||||
"name": "stdout",
|
||||
"text": "specific batch request https://13.91.32.53:30443/gateway/default/livy/v1/batches/1\n"
|
||||
},
|
||||
{
|
||||
"output_type": "stream",
|
||||
"name": "stdout",
|
||||
"text": "Response status is 200\n{'appId': None,\n 'appInfo': {'driverLogUrl': None, 'sparkUiUrl': None},\n 'id': 1,\n 'log': ['stdout: ',\n '19/06/18 17:57:32 WARN util.NativeCodeLoader: Unable to load '\n 'native-hadoop library for your platform... using builtin-java '\n 'classes where applicable',\n '\\nstderr: ',\n '\\nYARN Diagnostics: '],\n 'name': None,\n 'state': 'starting'}\nResponse received is None\n"
|
||||
}
|
||||
],
|
||||
"execution_count": 16
|
||||
}
|
||||
]
|
||||
}
|
||||
Binary file not shown.
|
After Width: | Height: | Size: 160 KiB |
@@ -1,81 +0,0 @@
|
||||
{
|
||||
"metadata": {
|
||||
"kernelspec": {
|
||||
"name": "pyspark3kernel",
|
||||
"display_name": "PySpark3"
|
||||
},
|
||||
"language_info": {
|
||||
"name": "pyspark3",
|
||||
"mimetype": "text/x-python",
|
||||
"codemirror_mode": {
|
||||
"name": "python",
|
||||
"version": 3
|
||||
},
|
||||
"pygments_lexer": "python3"
|
||||
}
|
||||
},
|
||||
"nbformat_minor": 2,
|
||||
"nbformat": 4,
|
||||
"cells": [
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"source": "# Read and write from Spark to SQL using the MSSQL jdbc Connector\r\nA typical big data scenario a key usage pattern is high volume, velocity and variety data processing in Spark followed with batch/streaming writes to SQL for access to LOB applications. These usage patterns greatly benefit from a connector that utilizes key SQL optimizations and provides an efficient and reliable write to SQLServer Big Data Cluster or SQL DB. \r\n\r\nMSSQL JDBC connector, referenced by the name com.microsoft.sqlserver.jdbc.spark, uses [SQL Server Bulk copy APIS](https://docs.microsoft.com/en-us/sql/connect/jdbc/using-bulk-copy-with-the-jdbc-driver?view=sql-server-2017#sqlserverbulkcopyoptions) to implement an efficient write to SQL Server. The connector is based on Spark Data source APIs and provides a familiar JDBC interface for access.\r\n\r\nThe following sample shows how to use the MSSQL JDBC Connector for writing and reading to/from a SQL Source. In this sample we' ll \r\n- Read a file from HDFS and do some basic processing \r\n- post that we'll write the dataframe to SQL server table using the MSSQL Connector. \r\n- Followed by the write we'll read using the MSSQLConnector.\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 spark_data and upload the file there. \r\n- Configure the spark session to use the MSSQL Connector jar. The jar can be found at /jar/spark-mssql-connector-assembly-1.0.0.jar post deployment of Big Data Cluster.\r\n\r\n``` sh\r\n %%configure -f\r\n {\"conf\": {\"spark.jars\": \"/jar/spark-mssql-connector-assembly-1.0.0.jar\"}}\r\n```\r\n\r\n \r\n ",
|
||||
"metadata": {}
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"source": "# Configure the notebook to use the MSSQL Spark connector\r\nThis step woould be removed in subsequent CTPs. As of CTP2.5 this step is required to point the spark session to the relevant jar.\r\n ",
|
||||
"metadata": {}
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"source": "%%configure -f\r\n{\"conf\": {\"spark.jars\": \"/jar/spark-mssql-connector-assembly-1.0.0.jar\"}}\r\n\r\n\r\n\r\n",
|
||||
"metadata": {},
|
||||
"outputs": [],
|
||||
"execution_count": 4
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"source": "# Read data into a data frame\r\nIn this step we read the data into a data frame and do some basic clearup steps. \r\n\r\n",
|
||||
"metadata": {}
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"source": "#Read a file and then write it to the SQL table\r\ndatafile = \"/spark_data/AdultCensusIncome.csv\"\r\ndf = spark.read.format('csv').options(header='true', inferSchema='true', ignoreLeadingWhiteSpace='true', ignoreTrailingWhiteSpace='true').load(datafile)\r\ndf.show(5)\r\n",
|
||||
"metadata": {},
|
||||
"outputs": [],
|
||||
"execution_count": 6
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"source": "\r\n#Process this data. Very simple data cleanup steps. Replacing \"-\" with \"_\" in column names\r\ncolumns_new = [col.replace(\"-\", \"_\") for col in df.columns]\r\ndf = df.toDF(*columns_new)\r\ndf.show(5)\r\n\r\n",
|
||||
"metadata": {},
|
||||
"outputs": [],
|
||||
"execution_count": 8
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"source": "# Write dataframe to SQL using MSSQL Spark Connector",
|
||||
"metadata": {}
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"source": "#Write from Spark to SQL table using MSSQL Spark Connector\r\nprint(\"Use MSSQL connector to write to master SQL instance \")\r\n\r\nservername = \"jdbc:sqlserver://master-0.master-svc\"\r\ndbname = \"MyTestDatabase\"\r\nurl = servername + \";\" + \"databaseName=\" + dbname + \";\"\r\n\r\ndbtable = \"dbo.AdultCensus\"\r\nuser = \"sa\"\r\npassword = \"****\" # Please specify password here\r\n\r\n\r\ntry:\r\n df.write \\\r\n .format(\"com.microsoft.sqlserver.jdbc.spark\") \\\r\n .mode(\"overwrite\") \\\r\n .option(\"url\", url) \\\r\n .option(\"dbtable\", dbtable) \\\r\n .option(\"user\", user) \\\r\n .option(\"password\", password)\\\r\n .save()\r\nexcept ValueError as error :\r\n print(\"MSSQL Connector write failed\", error)\r\n\r\nprint(\"MSSQL Connector write succeeded \")\r\n\r\n\r\n",
|
||||
"metadata": {},
|
||||
"outputs": [],
|
||||
"execution_count": 10
|
||||
},
|
||||
{
|
||||
"cell_type": "markdown",
|
||||
"source": "# Read SQL Table using MSSQL Spark connector.\r\nThe following code uses the connetor to read the tables. To confirm the write about check table directly using SQL",
|
||||
"metadata": {}
|
||||
},
|
||||
{
|
||||
"cell_type": "code",
|
||||
"source": "#Read from SQL table using MSSQ Connector\r\nprint(\"read data from SQL server table \")\r\njdbcDF = 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 .load()\r\n\r\njdbcDF.show(5)",
|
||||
"metadata": {},
|
||||
"outputs": [],
|
||||
"execution_count": 11
|
||||
}
|
||||
]
|
||||
}
|
||||
Reference in New Issue
Block a user