Skip to main content

Show us your Pipelines

Share an example of your Pipeline pattern. Upload a Screenshot and detailed description. No wrong answers. Do NOT share personal or credential information.

  • 68 Topics
  • 73 Replies
68 Topics
A
Anonymous
posted in  Show us your Pipelines
What's Your Favorite Dinosaur? - StreamSets Pipeline to Clean Delimited Files

Problem StatementYou’ve been asking everybody you meet the age old question: What is your favorite dinosaur? You’ve been saving their responses in a series of delimited files that have this format.  Tragically, the data you have meticulously gathered is dirty. Before you can execute your master plan to do whatever is that you were planning to do with this information, you must clean your dataset and resave it to a single clean file. The PipelineYou read in the Dino data with the Directory origin.  You remove duplicates on the Dinosaur column using the Record Deduplicator You filter Dinos that have murdered people in the Jurassic Park film series for reasons that are you own using the Stream Selector. Finally, you save your Dino data to one single file on your local system. Your data is clean, filtered and ready for analysis.  If you’re reading this, please comment with your favorite dinosaur. It’s uh...for science. 

621
Drew Kreiger
Rock star
2 years ago
kate
StreamSets Employee
kateStreamSets Employee
posted in  Show us your Pipelines
Build a Slowly Changing Dimension with Transformer for Snowflake

This pipeline is populating and updating a Slowly Changing Dimension for the NYC Citibike stations using Transformer for Snowflake. The pipeline is using StreamSets’ Slowly Changing Dimension stage that handles all of the logic for Type 1 and Type 2 dimensions. The SCD stage then produces the correct output along with the metadata that the Snowflake Destination uses to perform the merge.

680
kate
StreamSets Employee
2 years ago
kate
StreamSets Employee
kateStreamSets Employee
posted in  Show us your Pipelines
Using Transformer for Snowflake to create a fact table for Citibike Trips data

Transformer for Snowflake: Pipeline to Create Fact TableThis pipeline uses the NYC Citibike data to populate a fact table enriched with station and weather data.  Since this is using Transformer for Snowflake, none of the data leaves the Snowflake Data Cloud since the raw trips data is already loaded in Snowflake! The pipeline generates the SQL that is then executed on Snowflake, inserting the query results into the target tables. The pipeline is using weather data from Weather Source’s free Global Weather & Climate Data for BI data set on the Snowflake Data Marketplace so you can see the weather conditions for each ride.

490
kate
StreamSets Employee
2 years ago
kate
StreamSets Employee
kateStreamSets Employee
posted in  Show us your Pipelines
Load Real-Time 911 Data to Snowflake

 Read 911 Event Data and Load to SnowflakeThis pipeline uses the HTTP client to read real time 911 event data from the Seattle Fire Department. The pipeline then flattens a nested JSON object It also uses the latitude and longitude of the incident location to lookup the Census block information from the Census Bureau’s geocoding API.Fun fact: While building this pipeline, an event came through for a dumpster fire currently happening on my block !

400
kate
StreamSets Employee
2 years ago
T
StreamSets Employee
thomas.bennettStreamSets Employee
posted in  Show us your Pipelines
Reverse ETL (MySQL to MixPanel)

This pipeline performs a Reverse ETL process. We take the enriched, coalesced data and then generate the proper JSON payload that is sent to MixPanel to either create or update users.  

460
T
StreamSets Employee
2 years ago
subashini
StreamSets Employee
subashiniStreamSets Employee
posted in  Show us your Pipelines
Solving Invalid Cipher text issue with Kafka Stages using JSON Data format

Scenario : Processing data in JSON format using Encrypt/decrypt processor and loading it to a destination. Steps to implement the pipeline without receiving Invalid ciphertext type error.1.First Pipeline reads data in JSON format sends it to the Encrypt/decrypt processor with action as Encrypt and load it to Kafka Producer with data format as JSON.Few columns are encrypted using Amazon KMS and fed to Kafka producer.  2.Second pipeline reads data from Kafka Consumer with data format as JSON.Field Type convertor is used to converts columns to Byte Array before feeding data into Encrypt/Decrypt processor because the Encrypt/Decrypt processor will only accept columns with data type as Byte Array.Columns that are converted to Byte Array are sent one by one into a Base64 Field Decoder. Please note that We can pass one field per Base64 Field Decoder.From the Base64 Field Decoders the columns are sent to Encrypt/decrypt processor with action as decrypt and finally the data is written to a file

3140
subashini
StreamSets Employee
2 years ago
wilsonshamim
StreamSets Employee
wilsonshamimStreamSets Employee
posted in  Show us your Pipelines
IOT Usecase : Collect Device data to Databricks Delta lake/Elasticsearch

Use Case:  Receive Device data from various machines around the world either using (python scripts/Go lang/MQTT/Kafka/Files). All the data is landed on MQTT Topics. Streamsets pipeline will read the data from the MQTT Topics.Once device data is received from the MQTT Topics, will get more machine details (like Country/City/longitude/Latitude) using Geo IP.Raw data will be stored in S3.Processed Data will be stored into Databricks delta lake and elasticsearch for analysis. Sample analysis:Count all devices for a particular country and map them.Sample Data:battery_level,c02_level,cca2,cca3,cn,device_id,device_name,humidity,ip,latitude,lcd,longitude,scale,temp,timestamp1,234,US,USA,United States,1,meter-gauge-abcdef,10,111.161.225.1,38,green,-97,Celsius,14,1458444054093  Complete Pipeline:  

2340
wilsonshamim
StreamSets Employee
2 years ago
Bikram
Headliner
BikramHeadliner
posted in  Show us your Pipelines
Read multiple csv from S3,Handle corrupt data,Mask the sensitive data and send Data to on premise and Denodo DB

    This is a generic pipeline that has been created to  read multiple .csv files from AWS s3 bucket and handles corrupt data if any in the file and masks the data based on DB env and performs full and data load  to the destination DB’s.Let's deep drive into each and every processor which is being used in the pipeline.Origin : AWS S3 : Read multiple .csv files from the bucket.Processors:Expression Evaluator : This is used for setting the pipeline execution time and user to identify who executes the pipeline and when.Field Type Converter:  This is for converting time field  to correct date time format (YYYY-MM-DD HH:MM:SS)Jython Evaluator : This is for handling corrupt data if any in the file (e.g “‘ street address ’”)Stream-sets Selector : This is for checking  for which environment  data needs to be masked before loading data to destination DB’s. In this case for Dev and Test env data will be masked because of GDPR rule we couldn’t load sensitive data into it but for Prod environment

3010
Bikram
Headliner
2 years ago
onefoursix
StreamSets Employee
onefoursixStreamSets Employee
posted in  Show us your Pipelines
Kafka Stream Processor combined with Push-down ETL on Snowflake and Databricks Delta Lake

Here's a Kafka Stream Processor pipeline that reads events from a "raw" topic, performs streaming transforms, and publishes the transformed events to a "refined" topic that multiple downstream clients subscribe to: Kafka Stream Processor Pipeline Here is the Stream Processor Pipeline’s placement within a StreamSets Topology that allows visualization and monitoring of the end-to-end data flow, with last-mile pipelines moving the refined events into Delta Lake, Snowflake, Elasticsearch, S3 and ADLS:Kafka Stream Processor with Multiple Consumers Users can extend the Topology using Transformer for Snowflake to perform push-down ETL on Snowflake using Snowpark, and Transformer on Databricks Spark to perform ETL on Delta Lake Push-down ETL on Snowflake and Databricks Delta Lake    

1250
onefoursix
StreamSets Employee
2 years ago
Rishi
StreamSets Employee
RishiStreamSets Employee
posted in  Show us your Pipelines
Solving Fun Data problem with Streamsets transformer

In this pipeline we solve some fun data problem with the help of  StreamSets Transformer . Transformer execution engine runs data pipelines on Apache Spark. We can run this pipeline on any spark cluster type Problem Statement:  Find the average number of friends for each age and sort them in ascending order.We are given fake friend dataset of social networking platforms in CSV file format is stored in Google cloud storage id, Name, Age, Number of Friends0,Will,33,3851,Jean-Luc,26,22,Hugh,55,2213,Deanna,40,4654,Quark,68,215,Weyoun,59,3186,Gowron,37,2207,Will,54,307   

380
Rishi
StreamSets Employee
2 years ago
Rishi
StreamSets Employee
RishiStreamSets Employee
posted in  Show us your Pipelines
CDC pipeline from Mysql to Snowflake

 MySQL binary logs provide a very efficient way to track data changes. It contain events that describe the modifications to data. In short, binary logs contain all the information that is required to trace the global status of the server right from the time it was started. This pipeline is design to propagates INSERT, UPDATE, and DELETE operations from MySQL BinLog to Snowflake.

1481
Drew Kreiger
Rock star
2 years ago
antmcmullen
StreamSets Employee
antmcmullenStreamSets Employee
posted in  Show us your Pipelines
Ordinance Survey - Address Base Premium Processing mutli-schema documents easily

Ordnance Survey (OS) produce a suite of products for the UK Market, One of these is Address Base Premium (ABP), Which is a set of details that describe buildings, businesses, residential and items you’d find on a map in detail, such as Lat/Long data, how the royal mail refers to it, and how the local authorities/councils describe and classify those items.Sounds good right? Well its not so nice of a set of data to deal with, It is shipped in 5 km batches of data in a single csv file, across 10 different schema patterns within it.. Nightmare? Nope! You know the secret of streamsets!Streamsets doesn't care about schema on read, That is the key to unlocking this… In the ABP Files, the first column has a record identifier, This tells us which schema and rules to process.This pipeline makes such a difficult issue to deal with normally, clear and transparent.You can watch a record come in from the raw file, read it with no header, Look at the first column. Process that to a given lane, That l

650
antmcmullen
StreamSets Employee
2 years ago
antmcmullen
StreamSets Employee
antmcmullenStreamSets Employee
posted in  Show us your Pipelines
Split child records that are crosstabbed back into flat records

Data comes from a monitor device, with test results of different elements. Those elements have 3 values, their ID, Their Value and any error messages.The customer wants to see a flattened list of just Monitor Device and timestamp with a result of each element on seperate lines.Using the above we can achieve that, we import the data ignoring the header line. (hence labeling them with numbers) First we label the monitor “Parent” fields by using a field renamer We build a Empty map for us to correctly parse the records using expression evaluator  we use a field mapping processor, to map those groups of 3 fields ( checking we only remap columns that are numeric) now we have groups, we split the groups into records using a field pivot processor This leaves a single group per record but as a group, Lets tidy that up with a field flattener so all the records are at the same level  Finally we use a field renamer to label the 3 fields we have produced We ship that off to our secure storage faci

490
antmcmullen
StreamSets Employee
2 years ago
Dash
Headliner
DashSenior Technical Evangelist and Developer Advocate at Snowflake
posted in  Show us your Pipelines
Slowly but surely!

 This pipeline is designed to capture inserts and updates (SCD Type II) being uploaded to a bucket on Amazon S3 for a slowly changing dimension table -- Customers. The pipeline creates new records with version set to 1 for new customers and with version set to (current version + 1) for existing customers. The customer records are then stored in Snowflake Data Cloud.

500
Dash
Headliner
2 years ago
Dash
Headliner
DashSenior Technical Evangelist and Developer Advocate at Snowflake
posted in  Show us your Pipelines
Real-time scoring using ML model

 This pipeline is designed to ingest streaming data from Kafka and load a trained ML model in Scala custom processor to predict sentiment of tweets. The pipeline runs on Databricks cluster and stores the tweets along with its score in Delta Lake.

580
Dash
Headliner
2 years ago
Dash
Headliner
DashSenior Technical Evangelist and Developer Advocate at Snowflake
posted in  Show us your Pipelines
Train ML Model and register experiment in MLflow

 This pipeline is designed to ingest data from Amazon S3 and prepare it for training a ML model using PySpark custom processor. Once the Gradient Boosted model is trained, the model artifacts, features, accuracy of the model and other metrics are registered as an experiment in MLflow. (The pipeline runs on Databricks cluster which comes bundled with MLflow server.) 

470
Dash
Headliner
2 years ago
Dash
Headliner
DashSenior Technical Evangelist and Developer Advocate at Snowflake
posted in  Show us your Pipelines
Embrace Data Drift or get left behind!

 This pipeline is designed to handle (embrace!) data drift while ingesting web logs from AWS S3 and then transforming and enriching them before storing the curated data in Snowflake Data Cloud. The data drift alert is triggered if/when the data being ingested is missing a key field IP Address which is crucial for downstream analytics. 

380
Dash
Headliner
2 years ago

Badge winners

Show all badges
Terms & Conditions