Quickstart using CLI

Use GlassFlow CLI to create a pipeline to process incoming data, transform it, and make it available for consumption in real-time.

Prerequisites

To complete the Quickstart you'll need the following:

1. Python is installed on your machine.

2. Download and Install Pip to manage project packages.

Install GlassFlow CLI

Install the GlassFlow CLI on macOS using the Homebrew:

brew tap glassflow/tap && brew install glassflow

You can also install the CLI using other installation options on Linux/Windows.

Create a GlassFlow account

To initiate the signup process via the GlassFlow CLI, run the following command:

glassflow signup

Follow the sign-up steps, after successfully creating an account, you are ready to create a new pipeline.

Creating Your First Pipeline

Step 1: Setup Project

Start by creating a dedicated project folder. Create a directory for this project named glassflow-playground, where you will place all the necessary files.

mkdir glassflow-playground && cd glassflow-playground

Step 2: Install GlassFlow SDK

Install the GlassFlow Python SDK and virtual environment package python-dotenvusing pip.

Optional: Create a virtual environment before installing Python dependencies. Run the following command: python -m venv .venv && source .venv/bin/activate

pip install glassflow python-dotenv

Step 3: Create a Transformation Function

Create a new file transform.py and write a short Python code that transforms incoming data. The function you write will parse the incoming JSON data, convert product quantities based on unit measurements, and categorize product items.

https://github.com/glassflow/glassflow-examples/blob/main/docs/quickstart/transform.py

import json

# Write a mandatory 'handler' function
def handler(data, log):
    log.info("Event:" + json.dumps(data), data=data)
    # Example transformation: Convert units from pounds to kilograms
    if data['unit'] == 'pounds':
        pound_quantity = data['quantity']
        data['quantity'] = float(pound_quantity) * 0.453592       
        data['unit'] = 'kg'
    
    return data

Note that the handler function is mandatory to implement in your code. Without it, the running transformation function will not be successful.

You can also import other Python dependencies (packages) in the transformation function. See supported libraries with GlassFlow.

Step 4: Create Your Pipeline

Run the following command in a terminal to create your pipeline in GlassFlow using the CLI tool.

glassflow pipeline create inventorymanagement --function=transform.py

Once the pipeline is configured successfully, the pipeline becomes active immediately.

Important: Now the transformation function is deployed and running on GlassFlow’s Serverless Execution Engine in the cloud, and you do not need to configure or maintain any infrastructure on your side.

Step 5: Get Your Pipeline Credentials

Run the following command to retrieve Pipeline ID(Identifier) and Space ID:

glassflow pipeline list

You will get a message on the terminal with the Pipeline ID and Space ID:

ID                                   | Name                     | Space ID                             | Space Name     |
-------------------------------------------------------------------------------------------------------------------------
9c43996a-2702-4f33-a0c2-711b4d18a7f6 | inventorymanagement      | 07409e3b-bad6-414d-a69c-0f2ce5a14982 | main           |

To get the pipeline access token, run the List Access Tokens CLI command:

glassflow pipeline tokens {pipeline_id}

Name    | ID                                   | Token     |
------------------------------------------------------------
default | 84b61c1c-0a92-4c2d-a1de-591793b12c30 | **********|
------------------------------------------------------------

Note down the Pipeline ID, Space ID, and Access Token. You'll set them as environment variables in the upcoming section.

Step 6: Create an environment configuration file

Create a .env file in your project directory with the following content, replacing the placeholders with your actual GlassFlow pipeline credentials:

PIPELINE_ID=your_pipeline_id
SPACE_ID=your_space_id
PIPELINE_ACCESS_TOKEN=your_pipeline_access_token

Sending Data to Your Pipeline

Now you can start sending data to the pipeline. GlassFlow will automatically run your transformation function on each event entering the pipeline and make your transformed data available in milliseconds.

Create a new Python file in your project root directory called producer.py. Write a script to read auto-generated inventory data and send it to the GlassFlow pipeline. Copy and paste the following code to producer.py file:

https://github.com/glassflow/glassflow-examples/blob/main/docs/quickstart/producer.py

# producer.py

import glassflow
from dotenv import dotenv_values
import datetime
import random


def random_event_json(event_number=1):
    return {
        "product_id": 1000 + event_number,
        "product_name": random.choice(
            [
                "Blue T-Shirt",
                "Black Jeans",
                "Running Shoes",
                "Baseball Caps",
                "Socks",
                "Leather Belts",
                "Wallets",
                "Sunglasses",
            ]
        ),
        "quantity": random.randint(20, 500),
        "unit": random.choice(["kg", "pounds"]),
        "supplier_id": random.choice(
            ["supplier1", "supplier2", "supplier3", "supplier4", "supplier5"]
        ),
        "created_at": datetime.datetime.now().isoformat(),
    }


def send_data_to_pipeline():
    config = dotenv_values(".env")
    pipeline_id = config.get("PIPELINE_ID")
    space_id = config.get("SPACE_ID")
    token = config.get("PIPELINE_ACCESS_TOKEN")

    client = glassflow.GlassFlowClient()
    pipeline_client = client.pipeline_client(
        space_id=space_id, pipeline_id=pipeline_id, pipeline_access_token=token
    )

    counter = 0
    try:
        while True and counter < 1000:
            # Create a random event
            event = random_event_json(counter)
            counter += 1
            try:
                res = pipeline_client.publish(request_body=event)
                if res.status_code == 200:
                    print("Event published successfully")
                    print(event)
            except Exception as e:
                print("Error in publishing event")
                print(e)
    except KeyboardInterrupt:
        print("Exiting ")


if __name__ == "__main__":
    send_data_to_pipeline()

Run the above Python script with the following command in your terminal:

python producer.py

When you run the above script, on every event that enters the pipeline, GlassFlow will invoke the transformation function that you have defined in the inventorymanagement. In this case, it will run the transform.py file and transform every raw data you send to the pipeline.

Consuming Data from Your Pipeline

Create a new Python file called consumer.py. The consumer is responsible for pulling transformed data from the pipeline. It continuously checks for new data, processes it as needed, and acts upon the transformed information. Copy and paste the following code to consumer.py the file:

https://github.com/glassflow/glassflow-examples/blob/main/docs/quickstart/consumer.py

# consumer.py

import glassflow
from dotenv import dotenv_values
import json
import sys


def read_data_from_pipeline(file_path):
    config = dotenv_values(".env")
    pipeline_id = config.get("PIPELINE_ID")
    space_id = config.get("SPACE_ID")
    token = config.get("PIPELINE_ACCESS_TOKEN")

    client = glassflow.GlassFlowClient()
    pipeline_client = client.pipeline_client(
        space_id=space_id, pipeline_id=pipeline_id, pipeline_access_token=token
    )

    with open(file_path, "a+") as f:
        while True:
            try:
                # consume transfornmed data from the pipeline
                res = pipeline_client.consume()
                if res.status_code == 200:
                    # get the transformed data as json
                    data = res.body.event
                    print("Data consumed successfully")
                    print(data)
                    f.write(json.dumps(data) + "\n")
                    f.flush()
            except KeyboardInterrupt:
                print("Exiting program")
                sys.exit(0)


if __name__ == "__main__":
    read_data_from_pipeline("supplier_inventory.txt")

Run the above Python script consumer.py in a separate terminal window to see the output side-by-side:

python consumer.py

Now you activated the consumer side of the pipeline. This consumer.py script retrieves the processed data from the pipeline in real-time and you will see the results of data transformation written to a new local file called (supplier_inventory.txt).

You can watch the new transformations by running the following command on a different terminal window

tail -f supplier_inventory.txt

Congratulations! You've set up a real-time inventory management system pipeline using GlassFlow.

Conclusion

In this Quickstart you've learned the following:

• How to install GlassFlow and set up a new project.

• How to create a data pipeline using the GlassFlow CLI.

• How to implement transform function and process data in real-time.

• How to publish data into the pipeline.

• How to consume data from the pipeline.