Our primary role<\/span> in the collaboration was to investigate\u00a0Azure Databricks<\/a>\u00a0and other streaming alternatives that might meet their requirements and to recommend the best approach from a technical standpoint. However, we also aimed to integrate the systems with other Azure services and online OSS libraries that could support the sort of pipeline ZenCity needed.<\/p>\n As the project progressed, our team discovered that there are very few online open source examples that demonstrate building a CI\/CD pipeline for a Spark-based solution. And, to the best of our knowledge, none of the examples demonstrated a Databricks-based solution that can utilize its rich features. As a result, we decided to provide a CI\/CD sample and\u00a0address the challenges of continuous integration and continuous deployment for a Databricks-based solution.<\/p>\n This code story describes the challenges, solutions and technical details of the approach we decided to take.<\/p>\n In searching out the most suitable solution for ZenCity, we faced the following challenges:<\/p>\n For simplicity, the architecture diagram below describes a single workload, chosen as an example from several somewhat similar ones that we focused on.\nThis diagram depicts the processing of tweets from a Twitter feed and analyzing them.<\/p>\n <\/p>\n The above architecture uses Databricks notebooks (written in scala<\/a>) and Event Hubs<\/a>\u00a0to distinguish between computational blocks and enable scalability in a smart way.<\/p>\n The pipeline works as a stream that flows in the following manner:<\/p>\n Databricks\u00a0is a management layer on top of Spark that exposes a rich UI with a scaling mechanism (including REST API and cli tool) and a simplified development process. We chose Databricks specifically because it enables us to:<\/p>\n The option to create clusters on demand can also potentially enable a separate execution environment for a specific customer that scales according to their individual need.<\/p>\n * The idea of notebooks used in Databricks, is borrowed from Jupyter Notebooks<\/a>\u00a0and meant to provide an easy interface to manipulate queries and interact with data during development.<\/p><\/blockquote>\n In order to create a maintainable solution that supports CI\/CD and manual deployment with ease, it was necessary to have a suite of scripts that could support granular actions like “deploy resources to azure”<\/strong>\u00a0or\u00a0“upload environment secrets to Databricks.”<\/strong>\u00a0While almost all of the actions are achievable using azure-cli, Databricks-cli, or other libraries needed to deploy, build and test such a solution, it is essential to be able to call on those actions quickly when developing the solution and wanting to check changes. More importantly, it is critical for supporting a CI\/CD pipeline that doesn’t require any manual interaction.<\/p>\n To aggregate all scripts\/actions into a manageable and coherent collection of commands, we built upon Lace Lofranco’s work<\/a> and used make<\/a>\u00a0which can be run locally from a Linux terminal or on Travis.<\/p>\n Using make<\/strong>, the entire solution can be deployed by running\u00a0make deploy<\/span>\u00a0, while providing (according to prompt) the appropriate parameters for\u00a0the resource group name<\/strong>, region<\/strong> and subscription id<\/strong>.<\/p>\n The Makefile deployment script, runs a collection of script that uses azure-cli<\/a>, databricks-cli<\/a>\u00a0and Python scripts.<\/p>\n Using our sample project on GitHub,<\/a>\u00a0you can run deployment on Azure from your local environment and follow the prompt for any details.<\/p>\n Running the make deploy<\/span>\u00a0 command can also be done with all parameters, but would still require a prompt for a token from Databricks by using the command:<\/p>\n To set up a test environment, follow the Integration Tests<\/a>\u00a0section on the README file.<\/p>\n The ARM template enabled us to deploy all the resources in the solution in a single resource group, while associating the keys and secrets between them. Using ARM deployment, all resources, except for Databricks, could also be configured, and the various secrets and keys could be configured quickly.<\/p>\n We also used the ARM\u00a0output<\/strong> feature to export all keys and secrets into a separate\u00a0.env<\/strong> file which could later be used to configure Databricks.<\/p>\n To configure Databricks, we used\u00a0databricks-cli<\/strong>, which is a command line interface tool designed to provide easy remote access to Databricks and most of the API it offers.<\/p>\n The first script uploads all the relevant secrets into the Databricks environment, making them available to all clusters that will be created in it. The second script configures the libraries, clusters, and jobs that are required to run as part of the pipeline.<\/p>\n To make sure the notebooks run with test configuration, it is important to execute the notebooks with a declared parameter, setting the social source to “Custom”:<\/p>\n This line adds a parameter to the job execution of each notebook. The notebooks which are affected by this parameter will change their functionality to run with mock data.<\/p>\n When running on a test environment, it was necessary to remove the excess resources once the test had completed its execution. In a full scenario, we’d be able to completely delete the ARM resource group and re-create it in the next execution. But, because there’s currently no API for creating a token for Databricks, it was necessary to generate the token manually. This meant we were not able to delete the resources between executions and it was important to keep the Databricks resource (although not its clusters) alive between test runs. Because this was the case, we needed to make sure all jobs we initiated were terminated so that they didn’t continue to consume resources – the cleanup script would only<\/strong> be run\u00a0in a test deployment, after a successful \/ failed test run.<\/p>\n The build stage is used to build the Java packages that are uploaded and used in the job execution on Databricks.<\/p>\n The following test is run by Travis-CI to connect to Azure Event Hubs and listen on the last event hub in the pipeline, the event hub receiving the alerts.\u00a0If a new alert is identified, it will exit the test process successfully. Otherwise, it will fail the test.<\/p>\n Travis-CI<\/a> is a great tool for continuous integration, listening to GitHub changes, and running the appropriate deployment scripts.<\/p>\n In this project, we used Travis to listen to any change in the master branch and execute a test deployment. This configuration can also be changed to run once a day if you think that every change should\u00a0cause a build.<\/p>\n All the configuration of Azure and Databricks can currently be done remotely and automatically via the scripts described in this article, except for one task – creating an authentication token in Databricks.<\/a>\u00a0This task requires manual interaction with the Databricks API.<\/p>\n For that reason, to run tests in a test environment, it was necessary to first deploy a test environment, and use the output from the deployment to configure Travis repository settings with those parameters. To see how to do that, continue reading here<\/a>.<\/p>\n Together with ZenCity, we discussed adding Unit Testing to the Databricks pipeline notebooks, but we found that running unit tests on a stream-based pipeline was not something we wanted to invest in at that moment. For more information on Unit Testing Spark-based stream, see here<\/a>.<\/p>\n Integration testing on the other hand seemed like something that would help us figure out the maturity of the code being checked end to end, but also seemed quite a challenge. The idea behind the integration testing was to spin up an entire test environment, let the tests run with mock data, and “watch” the end of the pipeline while waiting for specific results.<\/p>\n We used Text Analytics<\/a> in Cognitive Services<\/a>\u00a0to get a quick analysis of the language and topics on each tweet. Although this solution worked great, the throttling limits on Cognitive Services (which are present on all SKU levels with varying limits) could not support a pipeline with the scale we were hoping to support.\u00a0Therefore, we used those services as an implementation example only, while in the customer deployment, we used scalable propriety models developed by the customer.<\/p>\n For that reason, in the published sample, we chose to mock those requests using a Function App<\/a>\u00a0with constant REST responses \u2013 in a real-life scenario this should be replaced with Cognitive Services (in case of a small scale stream), an external REST API, or a model that can be run by Spark.<\/p>\n In this sample, the production version uses Twitter API<\/a>\u00a0to read tweets on a certain hashtag\/topic and ingests them into the data pipeline. This API was encapsulated to enable mocking using predefined data in a test environment. Both the Twitter and the mock implementations use the following interface:<\/p>\n We started out looking to help ZenCity solve a challenge around building a cloud-based data pipeline, but found that the real challenge was finding and creating a CI\/CD pipeline that can support that kind of solution.<\/p>\n Along the way we worked on generalizing the solution in a way that would allow it to be configured and enhanced to work for any CI\/CD pipeline around a Databricks-based pipeline. The generalization process made our solution flexible and adaptable for other similar projects.<\/p>\n Additionally, the approach we took with the pipeline \u2013 which includes integration testing on the entire pipeline in a full test environment \u2013 can also be integrated into other projects with include a data pipeline.<\/p>\n There are two requirements needed to adopt this approach: first, a streaming pipeline, with which it’s hard to test each component or the SDK does not expose an easily testable API; and second, a data pipeline that enables controlling the input to the pipeline and monitoring the output.<\/p>\n The scripts in this solution can be changed to support the deployment and configuration of any remotely controlled streaming infrastructure.<\/p>\n This code story describes CSE’s work with ZenCity to create a data pipeline on Azure Databricks supported by a CI\/CD pipeline on TravisCI. The aim of the collaboration was to create a pipeline capable of processing a stream of social posts, analyzing them, and identifying trends.<\/p>\n","protected":false},"author":21371,"featured_media":10618,"comment_status":"open","ping_status":"closed","sticky":false,"template":"","format":"standard","meta":{"_acf_changed":false,"footnotes":""},"categories":[10,11,14,16],"tags":[60,67,76,124,144,166,177,291,320,333,334],"class_list":["post-9429","post","type-post","status-publish","format-standard","has-post-thumbnail","hentry","category-azure-app-services","category-big-data","category-cognitive-services","category-devops","tag-azure","tag-azure-cli","tag-azure-event-hubs","tag-cognitive-services","tag-databricks","tag-email","tag-featured","tag-pipelines","tag-scala","tag-spark","tag-spark-streaming"],"acf":[],"blog_post_summary":" This code story describes CSE’s work with ZenCity to create a data pipeline on Azure Databricks supported by a CI\/CD pipeline on TravisCI. The aim of the collaboration was to create a pipeline capable of processing a stream of social posts, analyzing them, and identifying trends.<\/p>\n","_links":{"self":[{"href":"https:\/\/devblogs.microsoft.com\/ise\/wp-json\/wp\/v2\/posts\/9429","targetHints":{"allow":["GET"]}}],"collection":[{"href":"https:\/\/devblogs.microsoft.com\/ise\/wp-json\/wp\/v2\/posts"}],"about":[{"href":"https:\/\/devblogs.microsoft.com\/ise\/wp-json\/wp\/v2\/types\/post"}],"author":[{"embeddable":true,"href":"https:\/\/devblogs.microsoft.com\/ise\/wp-json\/wp\/v2\/users\/21371"}],"replies":[{"embeddable":true,"href":"https:\/\/devblogs.microsoft.com\/ise\/wp-json\/wp\/v2\/comments?post=9429"}],"version-history":[{"count":0,"href":"https:\/\/devblogs.microsoft.com\/ise\/wp-json\/wp\/v2\/posts\/9429\/revisions"}],"wp:featuredmedia":[{"embeddable":true,"href":"https:\/\/devblogs.microsoft.com\/ise\/wp-json\/wp\/v2\/media\/10618"}],"wp:attachment":[{"href":"https:\/\/devblogs.microsoft.com\/ise\/wp-json\/wp\/v2\/media?parent=9429"}],"wp:term":[{"taxonomy":"category","embeddable":true,"href":"https:\/\/devblogs.microsoft.com\/ise\/wp-json\/wp\/v2\/categories?post=9429"},{"taxonomy":"post_tag","embeddable":true,"href":"https:\/\/devblogs.microsoft.com\/ise\/wp-json\/wp\/v2\/tags?post=9429"}],"curies":[{"name":"wp","href":"https:\/\/api.w.org\/{rel}","templated":true}]}}Challenges<\/span><\/h2>\n
\n
Solution<\/h2>\n
Architecture<\/h3>\n
![\"Image](\"https:\/\/devblogs.microsoft.com\/cse\/wp-content\/uploads\/sites\/55\/2018\/12\/ci-cd-pipeline-cloud-architecture.png\")
<\/p>\n\n
\n
Databricks<\/h3>\n
\n
Databricks Deployment Scripts<\/h2>\n
Getting Started<\/span><\/h3>\n
make deploy resource-group-name=sample-social-rg region=westeurope subscription-id=5b86ec85-0709-4021-b73c-7a089d413ff0<\/pre>\n
Deploying the ARM template<\/h3>\n
deploy\/deploy.sh [Shell]<\/h6>\n
# Deploying an ARM template with all resources to Azure\r\narm_output=$(az group deployment create \\\r\n --name \"$deploy_name\" \\\r\n --resource-group \"$rg_name\" \\\r\n --template-file \".\/azuredeploy.json\" \\\r\n --parameters @\".\/azuredeploy.parameters.json\" \\\r\n --output json)\r\n\r\n# Extracting deployment output parameters\r\nstorage_account_key=$(az storage account keys list \\\r\n --account-name $storage_account \\\r\n --resource-group $rg_name \\\r\n --output json |\r\n jq -r '.[0].value')\r\n\r\n# Dumping the secrets into a local .env file\r\necho \"BLOB_STORAGE_KEY=${storage_account_key}\" >> $env_file<\/pre>\nConfiguring Databricks Remotely<\/h3>\n
deploy\/databricks\/create_secrets.py [Python]<\/h6>\n
# Using Databricks REST API to create a secret for every environment variable in the .env file\r\napi_url = \"https:\/\/\" + dbi_domain + \"\/api\/2.0\/\"\r\nscope = \"storage_scope\"\r\n\r\nr = requests.post(api_url + 'preview\/secret\/secrets\/write',\r\n headers={\"Authorization\": \"Bearer \" + token},\r\n json={\"scope\": scope, \"key\": secret_name, \"string_value\": secret_value\r\n })\r\nresponse_body = r.json()\r\nif r.status_code != 200:\r\n raise Exception('Error creating scope: ' + json.dumps(response_body))\r\nreturn (response_body)<\/pre>\ndeploy\/databricks\/configure.sh [Shell]<\/h6>\n
# Configure databricks cli profile to use the user generated token\r\n> ~\/.databrickscfg\r\necho \"[DEFAULT]\" >> ~\/.databrickscfg\r\necho \"host = $DATABRICKS_URL\" >> ~\/.databrickscfg\r\necho \"token = $DATABRICKS_ACCESS_TOKEN\" >> ~\/.databrickscfg\r\necho \"\" >> ~\/.databrickscfg\r\n\r\n# Create + Start a cluster to use for library deployments\r\ndatabricks clusters create --json-file \".\/config\/cluster.config.json\"\r\ndatabricks clusters start --cluster-id $cluster_id\r\n\r\n# installing libraries on Databricks\r\ndatabricks libraries install --maven-coordinates com.microsoft.azure:azure-eventhubs-spark_2.11:2.3.1 --cluster-id $cluster_id\r\ndatabricks libraries install --maven-coordinates org.apache.bahir:spark-streaming-twitter_2.11:2.2.0 --cluster-id $cluster_id\r\ndatabricks libraries install --maven-coordinates org.json4s:json4s-native_2.11:3.5.4 --cluster-id $cluster_id\r\n\r\n# Uploading build artifacts from Java project (containing twitter wrapper and test validator)\r\nblob_file_name=\"social-source-wrapper-1.0-SNAPSHOT.jar\"\r\nblob_local_path=\"src\/social-source-wrapper\/target\/$blob_file_name\"\r\nblob_dbfs_path=\"dbfs:\/mnt\/jars\/$blob_file_name\"\r\n\r\ndatabricks fs cp --overwrite \"$blob_local_path\" \"$blob_dbfs_path\"\r\ndatabricks libraries install --cluster-id $cluster_id --jar \"$blob_dbfs_path\"\r\n\r\n# Uploading notebooks to Databricks\r\ndatabricks workspace import_dir \"..\/..\/notebooks\" \"\/notebooks\" --overwrite\r\n\r\n# Executing the jobs according to configuration paths\r\nfor filePath in $(ls -v $PWD\/config\/run.*.config.json); do\r\n declare jobjson=$(cat \"$filePath\")\r\n databricks runs submit --json \"$jobjson\"\r\ndone<\/pre>\n<\/div>\n<\/div>\n<\/div>\n
jobjson=$(echo \"$jobjson\" | \\\r\n jq '.notebook_task.base_parameters |= \\\r\n { \"socialSource\": \"CUSTOM\" }')<\/pre>\nCleanup<\/h3>\n
deploy\/databricks\/cleanup.sh [Shell]<\/h6>\n
# Stopping all active jobs deployed in the databricks deployment\r\nfor filePath in $(ls -v $PWD\/config\/run.*.config.json); do\r\n for jn in \"$(cat \"$filePath\" | jq -r \".run_name\")\"; do\r\n declare runids=$(databricks runs list --active-only --output JSON | \\\r\n jq -c \".runs \/\/ []\" | \\\r\n jq -c \"[.[] | \\\r\n select(.run_name == \\\"$jn\\\")]\" | \\\r\n jq .[].run_id)\r\n for id in $runids; do\r\n databricks runs cancel --run-id $id\r\n done\r\n done\r\ndone\r\n<\/pre>\n
Java Packages and Build<\/h3>\n
src\/integration-tests\/src\/main\/java\/com\/microsoft\/azure\/eventhubs\/checkstatus\/ReceiveByDateTime.java [Java]<\/h6>\n
package com.microsoft.azure.eventhubs.checkstatus;\r\n\r\n\/\/ Creating an Even Hub events receiver\r\nfinal EventHubClient ehClient = \r\n EventHubClient.createSync(connStr.toString(), executorService);\r\nfinal PartitionReceiver receiver = ehClient.createEpochReceiverSync(\r\n EventHubClient.DEFAULT_CONSUMER_GROUP_NAME, partitionId,\r\n EventPosition.fromEnqueuedTime(Instant.EPOCH), 2345);\r\n\r\nfinal LocalDateTime checkupStartTime = LocalDateTime.now();\r\n\r\n\/\/ Making sure 15 minutes haven't passed since the test started\r\nwhile (LocalDateTime.now().minusMinutes(15).isBefore(checkupStartTime)) {\r\n receiver.receive(100).thenAcceptAsync(receivedEvents -> {\r\n\r\n \/\/ After parsing each event, check event time is after start of execution\r\n if (eventDateTime.isAfter(startTime)) {\r\n System.out.println(\"Found a processed alert: \" + dataString);\r\n System.exit(0);\r\n }\r\n }, executorService).get();\r\n}\r\n\r\n\/\/ If no event was found for 15 minutes, fail the process\r\nSystem.exit(1);<\/pre>\nDatabricks Continuous Integration Using Travis<\/h3>\n
Integration Testing<\/h3>\n
Mocking Cognitive Services<\/h3>\n
Twitter API<\/h3>\n
src\/social-source-wrapper\/src\/main\/java\/social\/pipeline\/source\/SocialQuery.java [Java]<\/h6>\n
\/\/ Defining the interface to query a social service\r\npublic interface SocialSource {\r\n SocialQueryResult search(SocialQuery query) throws Exception;\r\n\r\n void setOAuthConsumer(String key, String secret);\r\n\r\n void setOAuthAccessToken(String accessToken, String tokenSecret);\r\n}<\/pre>\nConclusion<\/h2>\n
Resources<\/h2>\n
\n