Build Smarter AI Workflows with Datastreamer Pipelines

In data-driven decision making, converting raw data into insight you can act on is essential, yet the journey from ingestion to analysis is full of friction. For machine learning and innovation teams, building an AI data pipeline pulls resources away from core objectives. Datastreamer, a data pipeline platform, takes on the infrastructure concerns and provides the framework teams need to reach their analytical goals.

The problem: infrastructure slows AI teams down

Consider SocialMaple, a fictional company building machine learning models for trend prediction. Its data scientists want to focus on model development, while customers provide large datasets from cloud platforms such as Amazon S3 and Google Cloud Storage.

Before any prediction can run, raw data has to be ingested, transformed, enriched, and analyzed. Managing that end-to-end pipeline demands a significant infrastructure investment. It forces the team to divert resources from their primary mission of delivering real-time trend forecasts, and instead spend time maintaining complex backend systems.

The solution: an AI pipeline without the overhead

With Datastreamer, SocialMaple can build a complete pipeline in a handful of steps, keeping their attention on insights rather than backend management. The example pipeline has three parts:

• Google Cloud Storage Ingress ingests JSON files from a GCS bucket in batches.
• Custom Function processes each record through SocialMaple's prediction API, merging sentiment and named entity results back into the original JSON data.
• Datastreamer Searchable Storage stores the enriched, indexed data for fast querying.

Step 1: Google Cloud Storage ingress setup

Users add their Google Service Account credentials to API secrets, where they are encrypted and stored securely, then configure the ingress to pull JSON from the designated GCS buckets. The example uses fictional social media comments about non-alcoholic beer as sample data. Jobs monitor the buckets for incoming data, using folder scan options to check continuously for new files.

Step 2: data enrichment with a custom Python function

The pipeline enriches posts with a Python-based custom function that sends post content to an API. The SocialMaple API returns Named Entity Recognition (NER) results and sentiment labels for classification. The implementation looks like this:

import requests
import json

def process_batch(documents: list) -> list[dict[any, any]]:
    """
    Sends a request to social mapple prediction API and merges the response into the content
    """
    content_list = []
    for doc in documents:
        content_list.append(doc["content"]["body"])

    url = "https://socialmapple-prediction-344857442423.us-central1.run.app"

    headers = {
        'Content-Type': 'application/json'
    }

    response = requests.request("POST", url, headers=headers, data=json.dumps(content_list))
    predict_response = response.json()

    for i, doc in enumerate(documents):
        doc["enrichment"] = {
            "sentiment": predict_response[i]["sentiment"],
            "entity": predict_response[i]["entity"],
        }
        doc["data_source"] = "social_mapple"

    return documents

Step 3: searchable storage

The enriched JSON records ingest automatically into Datastreamer's internal searchable database. Data becomes query-ready the moment it enters, with no additional configuration required.

Deploying and querying the pipeline

Once the pipeline is configured and saved, users deploy it with the Deploy button. The Datastreamer Search API then queries the data, with extensible aggregation options. The example demonstrates a term aggregation on sentiment labels, filtered by the entity "non-alcoholic," and returns both the enriched content and the aggregated sentiment counts.

Data egress options

Datastreamer pipelines support multiple destination types:

Destination type	Examples
Analytics platforms	Snowflake, Databricks, BigQuery
Cloud storage services	GCS, Amazon S3, Azure
Custom workflows	Webhooks and Firehose delivery

Scaling AI workflows

This example shows Datastreamer simplifying a complete data pipeline, from Google Cloud Storage ingestion through a custom AI prediction function to searchable storage. Teams can enrich large data volumes, extract sentiment and entity insights, and make data available downstream without building complex infrastructure. By offloading backend processes, AI and data science teams keep their focus on model development and innovation, on a foundation that scales.

Datastreamer provides the tools to connect, enrich, and activate data without the infrastructure burden, whether the work involves real-time social data or training high-performance models.