Azure Stream Analytics example

azure-ts-stream-analytics/Pulumi.yaml

@@ -0,0 +1,10 @@
+name: azure-ts-stream-analytics
+runtime: nodejs
+description: Uses an Azure Stream Analytics job to transform data in an Event Hub.
+template:
+ config:
+ azure:environment:
+ description: The Azure environment to use (`public`, `usgovernment`, `german`, `china`)
+ default: public
+ azure:location:
+ description: The Azure location to deploy to (e.g., `eastus` or `westeurope`)

azure-ts-stream-analytics/README.md

@@ -0,0 +1,58 @@
+[![Deploy](https://get.pulumi.com/new/button.svg)](https://app.pulumi.com/new)
+
+# Azure Stream Analytics
+
+An example Pulumi program that deploys an Azure Stream Analytics job to transform data in an Event Hub.
+
+## Running the App
+
+1. Create a new stack:
+
+ ```
+ $ pulumi stack init dev
+ ```
+
+1. Login to Azure CLI (you will be prompted to do this during deployment if you forget this step):
+
+ ```
+ $ az login
+ ```
+
+1. Restore NPM dependencies:
+
+ ```
+ $ npm install
+ ```
+
+1. Configure the location to deploy the example to:
+
+ ```
+ $ pulumi config set azure:location <location>
+ ```
+
+1. Run `pulumi up` to preview and deploy changes:
+
+ ```
+ $ pulumi up
+ Previewing update (dev):
+ ...
+
+ Updating (dev):
+ ...
+ Resources:
+ + 15 created
+ Update duration: 2m43s
+ ```
+
+1. Use the following sample messages for testing:
+
+ ```
+ // Inputs (1 line - 1 event):
+ {"Make":"Kia","Sales":2,"Time":"2019-06-26T10:22:36Z"}
+ {"Make":"Kia","Sales":1,"Time":"2019-06-26T10:22:37Z"}
+ {"Make":"Honda","Sales":1,"Time":"2019-06-26T10:22:38Z"}
+
+ // Output:
+ [{"Make":"Kia","Sales":3};{"Make":"Honda","Sales":1}]
+
+ ```

azure-ts-stream-analytics/index.ts

@@ -0,0 +1,95 @@
+// Copyright 2016-2019, Pulumi Corporation. All rights reserved.
+
+import * as azure from "@pulumi/azure";
+
+// Create an Azure Resource Group
+const resourceGroup = new azure.core.ResourceGroup("streams-rg");
+
+// Define an Event Hub Namespace with two Hubs for an input and an output data streams
+const namespace = new azure.eventhub.EventHubNamespace("streams-ns", {
+ resourceGroupName: resourceGroup.name,
+ sku: "standard",
+});
+
+const inputHub = new azure.eventhub.EventHub("inputs", {
+ resourceGroupName: resourceGroup.name,
+ namespaceName: namespace.name,
+ partitionCount: 2,
+ messageRetention: 7,
+});
+
+const consumerGroup = new azure.eventhub.EventHubConsumerGroup("analytics", {
+ resourceGroupName: resourceGroup.name,
+ namespaceName: namespace.name,
+ eventhubName: inputHub.name,
+});
+
+const outputHub = new azure.eventhub.EventHub("outputs", {
+ resourceGroupName: resourceGroup.name,
+ namespaceName: namespace.name,
+ partitionCount: 2,
+ messageRetention: 7,
+});
+
+// Create a Stream Analytics Job that aggregates events per Make and 1-minute intervals
+const job = new azure.streamanalytics.Job("job", {
+ resourceGroupName: resourceGroup.name,
+ compatibilityLevel: "1.1",
+ dataLocale: "en-GB",
+ eventsLateArrivalMaxDelayInSeconds: 60,
+ eventsOutOfOrderMaxDelayInSeconds: 50,
+ eventsOutOfOrderPolicy: "Adjust",
+ outputErrorPolicy: "Drop",
+ streamingUnits: 1,
+ transformationQuery: `
+SELECT
+ Make,
+ SUM(Sales) AS Sales
+INTO
+ [MyOutput]
+FROM
+ [MyInput] TIMESTAMP BY Time
+GROUP BY
+ Make,
+ TumblingWindow(minute, 1)
+`
+});
+
+// Input of the job: the Event Hub with raw events
+const input = new azure.streamanalytics.StreamInputEventHub("input", {
+ name: "MyInput",
+ resourceGroupName: resourceGroup.name,
+ streamAnalyticsJobName: job.name,
+ servicebusNamespace: namespace.name,
+ eventhubName: inputHub.name,
+ eventhubConsumerGroupName: consumerGroup.name,
+ sharedAccessPolicyKey: namespace.defaultPrimaryKey,
+ sharedAccessPolicyName: "RootManageSharedAccessKey",
+ serialization: {
+ type: "Json",
+ encoding: "UTF8",
+ },
+});
+
+// Output of the job: the Event Hub with aggregated data
+const output = new azure.streamanalytics.OutputEventHub("output", {
+ name: "MyOutput",
+ resourceGroupName: resourceGroup.name,
+ streamAnalyticsJobName: job.name,
+ servicebusNamespace: namespace.name,
+ eventhubName: outputHub.name,
+ sharedAccessPolicyKey: namespace.defaultPrimaryKey,
+ sharedAccessPolicyName: "RootManageSharedAccessKey",
+ serialization: {
+ type: "Json",
+ encoding: "UTF8",
+ format: "Array",
+ },
+});
+
+// Create an Azure Function to subscribe to the output and print all outputs of the job
+outputHub.onEvent("analytics-output", {
+ callback: async (context, event) => {
+ console.log(JSON.stringify(event));
+ },
+});

azure-ts-stream-analytics/package.json

@@ -0,0 +1,11 @@
+{
+ "name": "azure-ts-stream-analytics",
+ "version": "0.1.0",
+ "devDependencies": {
+ "@types/node": "^10.3.3"
+ },
+ "dependencies": {
+ "@pulumi/azure": "latest",
+ "@pulumi/pulumi": "latest"
+ }
+}

azure-ts-stream-analytics/tsconfig.json

@@ -0,0 +1,22 @@
+{
+ "compilerOptions": {
+ "outDir": "bin",
+ "target": "es6",
+ "lib": [
+ "es6"
+ ],
+ "module": "commonjs",
+ "moduleResolution": "node",
+ "sourceMap": true,
+ "experimentalDecorators": true,
+ "pretty": true,
+ "noFallthroughCasesInSwitch": true,
+ "noImplicitAny": true,
+ "noImplicitReturns": true,
+ "forceConsistentCasingInFileNames": true,
+ "strictNullChecks": true
+ },
+ "files": [
+ "index.ts"
+ ]
+}