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.

  • 62 Topics
  • 71 Replies
62 Topics
realmatcha
Opening Band
realmatchaOpening Band
asked in  Show us your Pipelines
Connect SqlServer

Hi, I am new to Streamsets, started using it today.I have a task to create some backup data from sqlserver to hadoop and hive, the jdbc connection has been changed to sqlserver but it still doesn't work, please help me

4
realmatcha
Opening Band
1 year ago
HIMANSHU_SURANA
Fan
HIMANSHU_SURANAFan
asked in  Show us your Pipelines
Consuming MySQL binlog data in JDBC producer

I am trying to build CDC pipeline to migrate MySQL database to MySQL database in different server.Here’s the data collector pipeline I’ve created.  As per the documentation here, JDBC producers should be able to process binlog data. I’ve used Field remover to only use /Data and /Table fields.  When I run the pipeline I am getting error that input record has no data for <schema>.<table> . How can I create a pipeline to consume binlog records?

6
Bikram
Headliner
1 year ago
S
Fan
SubhaFan
asked in  Show us your Pipelines
HTTP Server

Can anyone let me know on how to use the HTTP Server Origin stage to fetch large volume of recordsMy problem statement is, I need to fetch large volume of data from a source using GET call. We can use HTTP Client stage for the same but since the source application doesn't have pagination configured , we will receive all the records at once . Thus i want to use HTTP server stage to achieve the same. Can anyone suggest how to achieve this. Appreciate your help!!Thanks in Advance!!

0
S
Fan
1 year ago
A
StreamSets Employee
ajinkyaStreamSets Employee
asked in  Show us your Pipelines
Sorting a specific column and writing it to a new table

Use Case:-We have a dataset, in which we have columns as follows:-FIRST_NAME, LAST_NAME, EMAIL, PHONE, GENDER, DEPARTMENT, JOB_TITLE, YEARS_OF_EXPEREIENCE, SALARY.  Lets sort the column SALARY in ascending order and write to a new table with just 4 columns, FIRST_NAME, LAST_NAME, YEARS_OF_EXPEREIENCE, SALARY. Pipeline Design:-Snowflake Table (origin) Sort (Processor) Column Remover (Processor) Snowflake Table (Destination) Pipeline Working:-Snowflake Origin will fetch the table and columns and pass the records to Sort processor Sort processor will sort the data based on the configuration and pass it to Column Remover. (SALARY column, Ascending order) Column Remover will keep or remove the columns based on the configuration. Snowflake Table Destination will write the data to a new snowflake table.  

0
A
StreamSets Employee
1 year ago
A
Discovered Fame
AnkurDiscovered Fame
asked in  Show us your Pipelines
Error: java.util.concurrent.TimeoutException: Idle timeout expired: 30000/30000 ms

Hi,We need to connect AWS S3 Select using Groovy scripting. For that need to upload jar files. While uploading below Jar files, getting the subject line error. The size of these jars is less than 1MB.joda-time.jarhttpclient.jarhttpcore.jaraws-java-sdk-s3.jarCan you please help me resolving this?Error: java.util.concurrent.TimeoutException: Idle timeout expired: 30000/30000 ms 

3
Bikram
Headliner
1 year ago
newkanyemerch32
Fan
newkanyemerch32Fan
asked in  Show us your Pipelines
btsmerchstore45

To operate on an EMR cluster, I have developed a straightforward transformer pipeline. I used the required configuration information during cluster configuration.I have the following problem: 

1
Bikram
Headliner
1 year ago
tamilarasup
Discovered Fame
tamilarasupDiscovered Fame
asked in  Show us your Pipelines
Transformer

In Transformer, i sent data from file(directory) to file(local fs)but the destination will got errorJOBRUNNER_63 - Pipeline status: 'RUN_ERROR', message: 'Operator File_3 failed due to org.apache.spark.sql.AnalysisException, check Driver Logs for further information'RUN_ERROR: Operator File_3 failed due to org.apache.spark.sql.AnalysisException, check Driver Logs for further information (View Stack Trace... ) how to clear the error and I need  help to clear the errorThanks Tamilarasu 

6
tamilarasup
Discovered Fame
1 year ago
tamilarasup
Discovered Fame
tamilarasupDiscovered Fame
asked in  Show us your Pipelines
Databricks Transformer

Hinow i am working in streamsets data-ops transformer platform. i try to connect databricks cluster in streamsets. But i facing some issues.  JOBRUNNER_63 - Pipeline status: 'START_ERROR', message: 'Failed to start pipeline, check logs for error. The Transformer Spark application in the Spark cluster might not be able to reach Transformer at ‘http://661f1312d1e9:19630’. If this is not the correct URL, update the transformer.base.http.url property in the Transformer configuration file or define a cluster callback URL for the pipeline and restart Transformer.'START_ERROR: Failed to start pipeline, check logs for error. The Transformer Spark application in the Spark cluster might not be able to reach Transformer at ‘http://661f1312d1e9:19630’. If this is not the correct URL, update the transformer.base.http.url property in the Transformer configuration file or define a cluster callback URL for the pipeline and restart Transformer. how to clear this error? and how to connect the cluster to

0
tamilarasup
Discovered Fame
1 year ago
tamilarasup
Discovered Fame
tamilarasupDiscovered Fame
asked in  Show us your Pipelines
mongodb

Is possible to perform cdc in mongodb atlas using mongodb oplog? The mongodb oplog connection string will be satisfied for MongoDB atlas but i got time out errorplease give some guidance to clear that error.

0
tamilarasup
Discovered Fame
1 year ago
Pradeep
StreamSets Employee
PradeepStreamSets Employee
posted in  Show us your Pipelines
Example pipelines for Delta lake lookup processor

 Input data:[  {    "product_id": 1,    "unit_price": 23.2  },  {    "product_id": 4,    "unit_price": 21.64  },  {    "product_id": 2,    "unit_price": 10.87  }]Delta lake lookup table. This is same for all operations performedReturn all matching rows, generating a record for each match:  Return first matching row:There won’t be change in the output from above since there are no duplicates in input. Return a count of matching rows:Return true if matches exist, otherwise false:Input data:[  {    "product_id": 1,    "unit_price": 23.2  },  {    "product_id": 2,    "unit_price": 21.64  },  {    "product_id": 4,    "unit_price": 10.87  },    {    "product_id": 5,    "unit_price": 99.32  },    {    "product_id": 4,    "unit_price": 20.33  }]Return all matching rows, generating a record for each match:Return first matching row:Return true if matches exist, otherwise false: Return a count of matching rows: 

0
Pradeep
StreamSets Employee
1 year ago
S
Discovered Fame
Santhosh KumarDiscovered Fame
asked in  Show us your Pipelines
what is best of streamsets?

Hello Team, My self Santhos kumari'm Beginner of streamsets so which one is best of the learning in depth of streamsets.

2
S
Discovered Fame
1 year ago
lakshmi_narayanan_t
Discovered Fame
lakshmi_narayanan_tDiscovered Fame
asked in  Show us your Pipelines
JDBC PIPE LINE NOT RUNING,SHOWS ME ERROR LIKE: JDBC_16 - Table '' does not exist or PDB is incorrect. Make sure the correct PDB was specified

I facing a issue on running JDBC pipeline ,it shows the below error ,I tried so many things  from the google but no use anybody can solve . 

2
lakshmi_narayanan_t
Discovered Fame
1 year ago
swayam
Discovered Fame
swayamDiscovered Fame
posted in  Show us your Pipelines
Convert multiple CSV records into a single JSON and make an HTTP bulk API POST call

Hi,The above pipeline reads the CSV files uploaded into an S3 bucket for new contact file uploads. Based on the new contacts available as a CSV, it converts them into a JSON. But the critical part here is that it merges each batches (each batch size is 1000) into a single JSON of multiple records and finally make a single HTTP POST call to one of our internal system to carry out a CONTACT update call.

5
R
Fan
1 year ago
Dash
Headliner
DashSenior Technical Evangelist and Developer Advocate at Snowflake
posted in  Show us your Pipelines
Orchestrate Initial Bulk Load to Change Data Capture

 This pipeline is designed to orchestrate initial bulk load to change data capture workload from on-prem Oracle database to Snowflake Data Cloud. The pipeline takes into consideration the completion status of the initial bulk load job before proceeding to the next (CDC) step and also sends success/failure notifications. The pipeline also takes advantage of the platform’s event framework to automatically stop the pipeline when there’s no more data to be ingested. 

1
P
Opening Band
2 years ago
G
Fan
gkognoleFan
asked in  Show us your Pipelines
Unable to ingest data from Azure SQL (CDC) to Azure Data bricks using Stream Sets.

Trying to build data pipeline for Azure SQL Server DB (CDC) as source and Azure Data bricks (Delta tables) as destinationI have referred data pipeline sample fromhttps://github.com/streamsets/pipeline-library/tree/master/datacollector/sample-pipelines/pipelines/SQLServer%20CDC%20to%20Delta%20Lake Getting below error for few records in Schema preview as-well:DELTA_LAKE_34 - Databricks Delta Lake load request failed: 'DELTA_LAKE_32 - Could not copy staged file 'sdc-4a076fce-7a73-45ba-8dd7-29e58848cf23.csv': java.sql.SQLException: [Simba][SparkJDBCDriver](500051) ERROR processing query/statement. Error Code: 0, SQL state: org.apache.hive.service.cli.HiveSQLException: Error running query: org.apache.spark.sql.AnalysisException: Unable to infer schema for CSV. It must be specified manually. Note : On Preview/Draft Run → Pipeline is able to capture changes from Source DB, successfully created files in stage (ADLS container) and created Delta tables at destination but it it fails to ingest re

4
G
Fan
2 years ago
iaakashbansal
Fan
iaakashbansalFan
asked in  Show us your Pipelines
HTTP Client Origin - Stop Condition for Pagination

I need to extract data from an API using OAuth2 connection. As per the data, they provide /cursor at the end of each page and that cursor can be used to get the records from next page.In Pagination tab, I used Link with Response field and tried to add Stop Condition with /cursor. but, not able to handle this scenario.Can someone please help. Thanks in Advance!!

1
Dimas Cabré i Chacón
StreamSets Employee
2 years ago
Drew Kreiger
Rock star
Drew KreigerSenior Community Builder at StreamSets
posted in  Show us your Pipelines
Announcement: Share your Pipeline Contest Winners! 🏆

Thank you to the 24 community members who submitted a pipeline! Our two winners for this quarters Share your Pipeline Contest is @swayam and @Bikram! (Stay on the look out for a DM from me.) Check out Swarm’s example pipeline, Carrying out image comparision using SDC pipeline and Amazon Image Rekognition. Where he compares the images received from facebook and twitter to determine if the 2 different profiles belongs to same customer or not. Bikram example Pipeline is, Read multiple csv from S3,Handle corrupt data,Mask the sensitive data and send Data to on premise and Denodo DB. His pipeline reads multiple .csv files from AWS s3 bucket and handles corrupt data if any in the file. The pipeline handles this while masking the data based on DB env and performs full data loads to the destination DB’s.  Stay on the look out for our next Share your Pipeline Contest! We are looking to use Miroverse for ease of trying new example pipeline via one click. 

5
S
Headliner
2 years ago
Drew Kreiger
Rock star
Drew KreigerSenior Community Builder at StreamSets
posted in  Show us your Pipelines
Share your Pipeline for a chance to win a $250 Amazon Gift Card! Example here!

Streamsetters!  2 Community members will each earn a $250 Amazon Gift card by submitting the most impressive or creative Pipelines. Please share your pipeline within this category (Show us your Pipeline).  It's easy to submit, Make sure to: Create a new Topic in the Show us your Pipeline category. (Each submission needs to have its own topic thread) Add an image of your sample pipeline. (MAKE SURE NO PERSONAL,CREDENTIALS, OR COMPANY INFO IS SCREEN SHOT OR SHARED) Add an explanation of 2-3 sentences of your pipeline use case and benefits. Lastly, Make it fun! Create something useful, fun, creative, and or wacky.   The closing date of our 1st “Share your Pipelines” is March 31, 2022 Example Entry:By @Mike Carley Here are other sample Pipelines. https://github.com/streamsets/pipeline-libraryHow we will choose the three winners: Winner will be announced on Mar 31, 2022.  The two winners will be graded on overall impressions, creativity, originality, and explantation. Winners will be

4
Drew Kreiger
Rock star
2 years ago
swayam
Discovered Fame
swayamDiscovered Fame
posted in  Show us your Pipelines
Handling Duplicate Data

Problem StatementVery often we find duplicate records being sent by customer while exporting data from their DWH and sending to us. To ensure data quality, we need to filter out the duplicate records before they enter into our platform for processing, computation and analysis.SolutionWe have addressed this using 2 separate pipelines.Pipeline-1: Files arriving in S3 gets picked up by this pipeline where for every record we calculate the MD5 Hash and make an entry in REDIS with KEY as “The MD5 hash value of the record” and VALUE as “Filename-RowNumber” of the current record being processed. Pipeline-2: The same file processed by Pipeline-1, again gets picked by Pipeline-2 and for every record it calculates the MD5 Hash and checks it’s value in REDIS to determine which RowNumber we need to process and which one we should drop as it’s a duplicate record.Pipeline-1:  Pipeline-1: Calculating Hash and Inserting to RedisAs you may see above:CSV file picked from S3 origin, gets converted to JSO

0
swayam
Discovered Fame
2 years ago
sundeep dhall
StreamSets Employee
sundeep dhallStreamSets Employee
posted in  Show us your Pipelines
Match and update incoming transactions in a Postgres database (JDBC, Postgres, AWS, Kafka)

 Streaming data pipeline that potentially has updates to transactions. Look up for matching records in the database (postgres), filter transaction using a condition to determine if transactions are updates vs new.  For updates, mask sensitive data and writes them to postgres db table.  New transaction are written to an AWS bucket and unprocessed transactions (pipeline errors) are written to a kafka destination for further analysis.

0
sundeep dhall
StreamSets Employee
2 years ago
swayam
Discovered Fame
swayamDiscovered Fame
posted in  Show us your Pipelines
Carrying out image comparision using SDC pipeline and Amazon Image Rekognition

We built this pipeline to compare the images received from facebook and twitter to determine if the 2 different profiles belongs to same customer ordifferent.We have a business case where we pull the social media comments from different users received from facebook and twitter. Apart from carrying out the sentiment analysis, we also need to know if same customer is being vocal in multiple platforms. Since customers normally don’t share their PII data, when we pull comments along with profile images from facebook and twitter, we store the images and consider the image matching as one of the criteria to determine the probability score of two comments received from different platfroms belongs to the same person or not.Before running the above pipeline, we have the pre-processed data with some probability score where the name, city, age, sentiments are already matched to come up with the best set of result to compare that is available in a database.In the above pipline,the origin is a MySQ

0
swayam
Discovered Fame
2 years ago
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. 

1
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.

0
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.

0
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 !

0
kate
StreamSets Employee
2 years ago

Badge winners

Show all badges
Terms & Conditions