Function to copy data from CSV and submit via function to Event Hubs
This project is a function app that collects CSVs from Blob and injects the rows one at a time into an Event Hub instance with optional delay. This is written in C# but you could achieve the same in other languages. See the video here
There are several pieces involved in writing a function app using triggers, inputs and outputs. You'll need to know about the following:
- Triggers
- Input Bindings
- Output Bindings
- Async apps
In addition to this, you'll need to configure the infrastructure, publish the function, and integrate the other components with bindings and app settings.
A trigger will be what causes your Function to run. An Input is optional, and used to collect data if your trigger does not provide it. Outputs are where your data is sent by the function. A full list of bindings can be found at functions-triggers-bindings and each one links to a full page of documentation explaining how to use that binding.
Some triggers will start an app and have no data, for instance a timer. Others will start an app with one piece of data. In this scenario, the app will do something with the data and then when the app is finished, the data is returned to the output binding and the app terminates. This is also the case with HTTP responses where a single response is required. Finally, you may have a trigger which provides multiple data in an array. In this instance you cannot simply return the data and terminate the app since you need to process all of the data. Here you should configure the app as async and return the data piece by piece until you run out and only then terminate the app. Each trigger, input and output will detail the ways of using it, but you may need to look carefully to notice the async keyword.
This example from the Event Hub documentation shows the return method if you have one output:
[FunctionName("EventHubOutput")]
[return: EventHub("outputEventHubMessage", Connection = "EventHubConnectionAppSetting")]
public static string Run([TimerTrigger("0 */5 * * * *")] TimerInfo myTimer, ILogger log)
{
log.LogInformation($"C# Timer trigger function executed at: {DateTime.Now}");
return $"{DateTime.Now}";
}
And this example from the Event Hub documentation shows the async method where you use an IASyncCollector object to iterate through data:
[FunctionName("EH2EH")]
public static async Task Run(
[EventHubTrigger("source", Connection = "EventHubConnectionAppSetting")] EventData[] events,
[EventHub("dest", Connection = "EventHubConnectionAppSetting")]IAsyncCollector<string> outputEvents,
ILogger log)
{
foreach (EventData eventData in events)
{
// do some processing:
var myProcessedEvent = DoSomething(eventData);
// then send the message
await outputEvents.AddAsync(JsonConvert.SerializeObject(myProcessedEvent));
}
}
Note the "public static async Task" which will also need System.Threading.Tasks to be added to the code to support the methods used.
This demo will require the following components:
- Function App
- Storage Account (Blob Containers)
- Event Hub
Optionally, you may also want to use Stream Analytics and Power BI to visualise the data, but I won't go into those here. You'll be able to see data hitting the Event Hub and can use Service Bus Explorer to view the messages on the bus.
In the Azure portal, create a new Function App
Create a new resource group for the demo called functionAppDemo and give your function a name such as csvToEH, you may want to add the date to make it unique. Select .NET Core as the runtime for this demo.
On the hosting tab leave the defaults in place for this demo, you may change these in production.
Click review and create and then create to complete setup.
In the Azure portal, create a new Storage Account
Select the same FunctionAppDemo resource group, give the storage account a name such as csv followed by the date, then select LRS for the storage tier. Click review and create then create.
Once the storage account has finished deploying, open it and click Containers.
Click add container and create a container called csvblobs. This container will be where we upload CSV files later to send through the function app.
Finally, click Access Keys on your storage account and copy the connection string
This will be used as an app setting in the function app later.
In the Azure Portal, create a new Event Hubs instance
Name the Event Hubs namespace and choose standard pricing tier. Select the same resource group and region.
Click Create.
Once deployed, open your event hub and click new Event Hub
Give the hub a name such as csvtoEH and click create
Now click Shared Access Policies and click Add
Call the policy functionApp so that you remember what you are granting access to, then give the policy send permission.
Once created, click the policy and copy the connection string. This will be used as an app setting for the Function App later on.
Now that you've configured the storage and event hubs, open your Function App and go to Configuration
Click on New Application Setting
Configure a setting called csvblobstore and paste your storage account connection string from above in the value
Now create a setting called eventhubconnection and paste your event hub connection string from above
These settings are explained below in the function.json section but are used to connect to the services.
Open Visual Studio and select Create New Project
Search for and select a new Azure Functions project using C#
Give your project a name and location and click create
Select Blob as your trigger and choose none for storage Account since we'll use the Azure Storage account created earlier. Name your connection string setting csvBlobStore and your path csvblobs, the same as the container you created. Click create to create the project
Once the project loads, right click on the project name in solution explorer and choose Manage NuGet Packages
Search for and install the Microsoft.Azure.WebJobs.Extensions.EventHubs package to install the Event Hubs binding
This file simply lists the version and sets up the Event Hub extension to be used by the app as per the documentation. It's pretty self explanatory but is necessary to be added or the extension is not recognised. This file is for the whole project so you only need to do this once for all of the included functions you create.
{
"version": "2.0",
"extensions": {
"eventHubs": {
"batchCheckpointFrequency": 5,
"eventProcessorOptions": {
"maxBatchSize": 256,
"prefetchCount": 512
}
}
}
}
This file is not in the project and is not editable. There will be one for each of your functions. It's created at build time by the compiler, based on C# attributes in the code (the parts in square brackets). An example is included here for reference. As you can see, this includes the blob trigger created with the project. It does not include the output binding, but instead specifies that this will be taken from attributes (code in square brackets) at runtime.
{
"generatedBy": "Microsoft.NET.Sdk.Functions-1.0.29",
"configurationSource": "attributes",
"bindings": [
{
"type": "blobTrigger",
"connection": "csvblobstore",
"path": "csvblobs/{name}",
"name": "myBlob"
}
],
"disabled": false,
"scriptFile": "../bin/CSVBlobToEH.dll",
"entryPoint": "CSVBlobToEH.Function1.Run"
}
From our source code, we have the input attribute:
[BlobTrigger("csvblobs/{name}", Connection = "csvblobstore")]Stream myBlob
This is what sets up the binding in the function.json file, and you can see the names will match, with the connection string csvblobstore being the name of the app setting we must use to configure the Blob connection string.
Next, we have the following output attribute:
[EventHub("dest", Connection = "eventhubconnection")]IAsyncCollector<string> outputEvents
This is where the app setting for the Event Hub is configured, so we need to create an app setting called "eventhubconnection" with the connection string in it. This is not stored in the function.json file, but is read directly by the app instead.
These settings are the ones you configured earlier in the function app on the portal.
This file simply overrides settings for use locally. Think of this as the location you configure application settings for debugging the code. Set these to the same values as the application settings you configured earlier.
{
"IsEncrypted": false,
"Values": {
"csvblobstore": "<enter your CSV source blob connection string here>",
"eventhubconnection": "<enter your event hub connection string here",
"FUNCTIONS_WORKER_RUNTIME": "dotnet"
}
}
At the top of the function, we set up the trigger, in this case BlobTrigger, using a connection with the name csvblobstore. This connection is configured either in your local settings if testing locally, or in the Function App settings in Azure. We set this up as a Stream variable named myBlob which means we can just treat it as a file in C#. We also set up the Event Hub output as an IAsyncCollector which we can write output events to. The whole function is async here as we will be sending multiple outputs from one input.
public static class Function1
{
[FunctionName("Function1")]
public static async Task Run([BlobTrigger("csvblobs/{name}", Connection = "csvblobstore")]Stream myBlob, [EventHub("dest", Connection = "eventhubconnection")]IAsyncCollector<string> outputEvents, string name, ILogger log)
Next, we write to the log some useful information. This is straight from sample code and not necessary but is recommended for troubleshooting. If you're wanting to tweak for performance you may want to leave this out if you're very confident in the app code and happy to use other methods for troubleshooting. We then set up a StreamReader to read the Blob line by line to process the CSV file. Please note that I included no code at all to check that we have a valid CSV file. This demo is all about setting up a function app so I left out everything but that.
{
log.LogInformation($"C# Blob trigger function Processed blob\n Name:{name} \n Size: {myBlob.Length} Bytes");
using (var reader = new StreamReader(myBlob))
{
Next, we read the first line and assume it's a header. If it wasn't, the JSON would use the first row as headers regardless and end up with some weird attribute names. Again, this is a minimum viable product to show Function Apps. We split the line by comma and create an array.
//Read the first line of the CSV file and break into header values
var line = reader.ReadLine();
log.LogInformation("headers: " + line);
var headers = line.Split(',');
Because it's useful to have a stream of events last for a while I added a delay, defined here. If you don't do this Event Hubs will just suck up all of the rows immediately (it's very capable!) so you won't get ongoing events for testing of other technologies. I made this app for testing events in Stream Analytics or PowerBI or some other Event Hub driven thing, maybe CosmosDB too.
//Define the sleep interval in milliseconds, use 0 if you don't want to wait, or remove the sleep command below
int delayTime = 1000;
Next, we do a while loop to read the remainder of the file line by line and make another array of the values. Since we want the rows to be sent as a JSON one at a time to create many events from one file, we set up an empty string. There are certainly ways to use Newtonsoft functionality to create JSON from an object, but since CSV is 2D data it's easy enough to manually create it. For each line, we add the current time to simulate a transaction time and then we also add the columns and values one at a time in a for loop. The final value can't have a comma in JSON so we add it outside the loop without the comma.
//Read the rest of the file
while (!reader.EndOfStream)
{
//Create an empty string for our JSON
string outputJSON = "";
outputJSON = outputJSON + "{\n";
//Read our lines one by one and split into values
line = reader.ReadLine();
var values = line.Split(',');
log.LogInformation("Values: " + line);
//Add a datestamp to the data
outputJSON = outputJSON + " functionDateStamp: \"" + DateTime.Now.ToString() + "\",\n";
//Add all of the data except the last value
for (int i = 0; i < (values.Length - 1); i++)
{
outputJSON = outputJSON + " \"" + headers[i] + "\": \"" + values[i] + "\",\n";
}
//Add the last value without a comma to properly form the JSON
int j = values.Length - 1;
outputJSON = outputJSON + " \"" + headers[j] + "\": \"" + values[j] + "\"\n";
//Close the JSON
outputJSON = outputJSON + "}";
Finally we add outputJSON, the string we created to the outputEvents variable, which is our output binding to Event Hubs. We do this with an await command to spawn a new thread task to avoid delaying the next line being processed. The main thread then carries on and reads the next line without delay. This makes the app extremely performant.
await outputEvents.AddAsync(outputJSON);
log.LogInformation("Added: " + outputJSON);
We then sleep the main thread (optionally) to spread the events out. Don't put this in a real Function as it does nothing useful in the real world. Event Hubs will easily accept your events at whatever speed you supply them!
//Sleep the function to delay the next event (otherwise it's really fast!)
System.Threading.Thread.Sleep(delayTime);
}
}
}
}
Right click on the project under Solution Explorer and select Publish
Select "Select Existing" to browse for your function app
Select your function app from the list and click OK
Finally, click Publish to push your code to the Azure service
To test the app, create a file in Excel and set up several columns with hundreds of rows or use the file here. Upload this to your blob container and the function app will find it and start pushing the rows to your Event Hubs instance as individual events.