Create a pipeline

This page explains how to create a new data pipeline in GlassFlow.

GlassFlow simplifies the creation and management of data pipelines by offering CLI-based and UI-based approaches for setup, run and deployment.

A set of tutorials under the Develop section demonstrates setting up a pipeline with a custom Python function, sending real-time data, and making the data available for external systems.

Mobility project example pipeline

We introduce a practical example of configuring a new data pipeline for a car-sharing (Mobility) project of companies like Bolt or Lyft. The mobility project aims to process real-time ride events from API to monitor fuel consumption and manage the supply of cars available for sharing. The primary objectives include identifying low-fuel vehicles and finding the nearest fuel station based on GPS coordinates and fuel type. It also discounts drivers based on their vehicle's fuel level and proximity to the nearest fuel station.

GlassFlow car sharing project data pipeline example

The full source code for the car-sharing project can be found on GitHub.

We'll use the GlassFlow CLI to create a new space and configure our data pipeline.

Prerequisites

We assume that you have already had the following before proceeding with the tutorial:

• You created a GlassFlow account.

• You installed GlassFlow CLI and logged into your account via the CLI.

• You have cloned the glassflow-examples GitHub repo locally.

Custom Transformation Function

For the mobility project, the function processes real-time ride event data to identify a vehicle with low fuel levels, find the closest fuel station, and calculate discounts for users who need to refuel. It enables the ride-sharing service to encourage drivers to refill vehicles when necessary and optimize fleet management.

The full source code for the car-sharing project can be found on GitHub.

Creating a transform function

The transformation logic is encapsulated within a mandatory Python function called handler, which processes each data event.

Create a Python script file transform.pyinside the mobility folder. The source code can also be found on the glassflow-examples repo on GitHub

https://github.com/glassflow/glassflow-examples/blob/main/tutorials/mobility/transform.py

"""
Transform function by the user
"""
import json
import requests


def handler(data, log):
    log.info("Event:" + json.dumps(data), data=data)
    try:
        transformed = handle(data)
    except Exception as e:
        log.error("Error in transformation", error=str(e))
        raise e
    return transformed


def get_nearest_fuel_station(gps_cordinates, fuel_type):
    print("get nearest fuel station")
    url = "https://mock-mobility-s3r3lbzina-ey.a.run.app/mobility/gas-stations/nearest"
    resp = requests.get(url,
                        params={
                            'cordinates_lat': gps_cordinates[0],
                            'cordinates_long': gps_cordinates[1],
                            'fuel_type': fuel_type
                        })
    if resp.status_code == 200:
        fuel_station = resp.json()
        print(fuel_station)
        return fuel_station
    else:
        print("error getting nearest fuel station")
        print(resp.status_code)
        return None


def handle(data: json):
    data['discount'] = {"discount": False}
    if not data['is_electric'] and data['current_fuel_percentage'] < 25:
        # find nearest gas station using a partner API
        fuel_station = get_nearest_fuel_station(data['gps_cordinates'],
                                                data['fuel_type'])
        if fuel_station:
            data['discount'] = {
                "discount": True,
                "fuel_station": fuel_station,
                "discount_type": "fuel"
            }

    return data

Thehandler function contains all transformation logic where the event data is modified based on specific conditions.

• If the vehicle is not electric and its current fuel percentage is below 25%, it calls the get_nearest_fuel_station function to find the nearest fuel station via the mock API server.

• If a fuel station is found, it updates the 'discount' key with details about the discount offered by the fuel station.

In the next steps, we will configure a pipeline on Glassflow with the transform.py created above.

Creating the pipeline on GlassFlow

Step 1: Create a new space

Open a terminal and create a new space called examples to organize multiple pipelines:

glassflow space create examples

After the space is created successfully, you will get a SpaceID in the terminal.

Save the SpaceID for reference. You'll set it as environment variables for the project in the upcoming section.

Step 2: Configuring the Pipeline

Create a new pipeline in the selected space with a transformation function:

glassflow pipeline create mobilitydemo --space-id={space_id} --function=transform.py

This command initializes the pipeline with a name mobilitydemo in the examples space and specifies the transformation function transform.py. After running the command, it returns a new Pipeline ID with its access token.

The pipeline is now deployed and running on the GlassFlow cloud. In the next sections, we will see how to publish data to this pipeline

Save the Pipeline ID and Access Token for reference. You'll set them as environment variables in the upcoming section.

Step 3: Create an environment configuration file

Create a new folder called mobility for the project and add a .env file in the project directory and add the following configuration variables:

PIPELINE_ID=your_pipeline_id
SPACE_ID=your_space_id
PIPELINE_ACCESS_TOKEN=your_pipeline_access_token

Replace your_pipeline_id, your_space_id, and your_pipeline_access_token with appropriate values obtained in the previous steps.

Next

We will publish data to the pipeline and consume the transformed data back in real-time