Skip to main content

StreamSets Platform

Get inspired and gain all the knowledge you need

1221 Topics
john.durkin
StreamSets Employee
john.durkinStreamSets Employee
posted in StreamSets Academy
"DataOps Platform Fundamentals" - questions from the video course

Hi -Welcome to the StreamSets Academy area of Community!Please post questions related to the self-paced video course “DataOps Platform Fundamentals”. Please use the tag “DataOps Platform Fundamentals Course” when creating your question under the “StreamSets Academy” category. Thank you! 

4080
john.durkin
StreamSets Employee
elefthkam
Fan
elefthkamFan
posted in Share your Best Practices
multiple pipelines writing to same table

I'm working with StreamSets and have three pipelines that each consume from different Kafka topics. Each pipeline writes to its own dedicated table (e.g., table1,table2, table3), but all three also write to a shared table . I'm using logic to avoid duplicates(update or insert). My concern is about concurrent writes — if all three pipelines try to write to the same row (shared table) at the same time, will StreamSets or the database handle it safely? I'm also wondering how to properly configure retry attempts, error handling, and backpressure in StreamSets to ensure that no data is lost or skipped if database locks or contention occur. What are best practices for configuring the JDBC  and pipeline error handling for this kind of multi-pipeline write scenario?

170
elefthkam
Fan
Sperch
Opening Band
SperchOpening Band
asked in Community Articles and Got a Question?
HTTP Client Loops Unceasingly

We have an HTTP Client origin that pulls from our ServiceNow site records that would be returned if we ran a report in ServiceNow directly. The records pulled appear at first blush to all be accounted for, but unfortunately the origin just keeps looping and pulling the every record over and over. I reduced the batch size to as low as 100 records without effect (i.e., it still loops endlessly).The URL with parameters is:https://<intentionally_redacted>.servicenowservices.com/api/now/table/task?sysparm_query=closed_atISNOTEMPTY%5Eclosed_at%3E%3Djavascript%3Ags.dateGenerate('2023-01-01'%2C'00%3A00%3A00')%5Eclosed_by.department.nameSTARTSWITHPBO%20-%20ETS%5Eassignment_group!%3Daeb1d3bc3772310057c29da543990ea2%5Eassignment_group!%3D4660e3fc3772310057c29da543990e0b%5EnumberNOT%20LIKEGAPRV%5EnumberNOT%20LIKERCC%5Esysparm_display_value=true%5Esysparm_limit=100%5Esysparm_offset=0I have the stage set to pull in Batch mode, and for pagination I have tried all 5 modes including “None”. Since

50
Sperch
Opening Band
AkshayJadhav
StreamSets Employee
AkshayJadhavStreamSets Employee
posted in StreamSets Academy
Credentials time out when using Start Control Hub jobs stage in the Data Collector pipeline.

Environment:StreamSets Data Collector v4.3. Control Hub v3.51.4. (Inactivity period for session termination = 30 Min) Issue:It has been observed that credentials time out is happening on the long running Streaming Pipelines when using Start Jobs stage. This can happen when using the User & Password Authentication type.The following error captured from control hub logs indicates that the token has expired. START_JOB_04 - Failed to start job template for job ID: START_JOB_04 - Failed to start job template for job ID: 1238759hvbv949:Test, status code '401': {"ISSUES":[{"code":"SSO_01","message":"User not authenticated"}]}, status code } Solution:After an internal investigation, found that  we are encountering COLLECTOR-1125, which has been addressed and resolved in data collector v5.1.

271
PurnaTalluri
Fan
Sperch
Opening Band
SperchOpening Band
asked in Community Articles and Got a Question?
Possible to query the SCH for the historically longest running jobs?

Is it possible to query the SCH for the longest running jobs? For example, say you have 100 jobs, and you want to know the 10 longest running jobs out of the total (I’d happily take just reporting on averages if that is all there is). It seems like this would be a useful report or API endpoint to have.

30
Sperch
Opening Band
SebastienM
Roadie
SebastienMRoadie
asked in StreamSets Academy
Gettting error zomato directory does not exist

I’m following the DataOps Platform Fundamentals course and the Build a Pipeline chapter has you enter “/zomato” as a files directory in the configuration of a Directory source.However, when validating the pipeline I get the error: SPOOLDIR_12-Directory ‘/zomato’ does not exist: conf.spoolDir.Any solution for this?

120110
Ankit.Kumar53
Fan
E
Fan
Eric MencariniFan
asked in Community Articles and Got a Question?
Retreaving more metrics fom Jobs(SDK)

Hello!!Currently, I’m using the SDK to extract metrics from some jobs. However, instead of having only {run_count, input_count, output_count, total_error_count} from sch.job.metrics, I would like to have some other metrics as well, such as counting each record for each step of my job. Is this possible? How can I achieve this?I’m also looking to get the last received record (for each pipeline) that we can see in the real-time summary. How can I get this metric from ControlHub using the SDK?

321
palmeida123
Fan
S
Fan
Sri25Fan
asked in Community Articles and Got a Question?
Error while testing a connection and connecting to a SFTP server

Hi ,I am creating a new Connection  with SFTP protocol and trying to connect to the SFTP server using a private key .I inputted these while creating the Connection:Authentication as Private Key ,Private Key provider as plain text Private Key as the text that i copied from the ppk file Username the correct one there was no Passphraseand when i hit Test Connection I get the below errorStage 'com_streamsets_pipeline_lib_remote_RemoteConnectionVerifier_01' initialization error: java.lang.NullPointerException: Cannot invoke "net.schmizz.sshj.userauth.keyprovider.KeyProvider.getPublic()" because "this.kProv" is null (CONTAINER_0701) But alternatively if i input the same in the credentials tab of the pipeline and preview ,I am able to successfully connect to the sftp server and read the file there . Problem is when i create the same via a connection Kindly help me to resolve this issue .

150
S
Fan
Sperch
Opening Band
SperchOpening Band
asked in Community Articles and Got a Question?
Configuration of Postman to Use StreamSets APIs

I am attempting to use Postman to query the runtime history of all jobs so that I can provide a list of the X slowest running jobs. I have been unable to configure Postman correct for this GET call, and would like to know from the hivemind what I am doing wrong here.Trying to use this API endpoint:https://{Control_Hub_URL}.streamsets.com/jobrunner/rest/v1/job/{jobId}/historyAuth is set to “No auth” based upon another forum post I read that stated we should include the ID and key for the API credentials used in the header, though I have tried it with API Key and providing the credentials there, as well.I have read through the 14 questions and 4 conversations that return when searching here for “Postman” and tried a few of the proffered solutions but without success. I have no parameters and the headers are where I have been trying to define API credentials. Note these credentials are used by automation in the Control Hub so they are valid. I think my issue lies in what header names I ne

150
Sperch
Opening Band
mithunvs47
Fan
mithunvs47Fan
asked in Community Articles and Got a Question?
Assigning value to runtime parameter in pipeline start event as jdbc query

I am using Data collector 5.8.1 . I am selected JDBC in my pipeline start event I am using this query in my start event: DECLARE @XYZWorkFlowID INT = (SELECT TOP 1 WorkFlowID FROM dbo.RefWorkFlow WHERE WorkFlowName = 'XYZ')INSERT INTO JobTracker(JobName,WorkFlowID,Campaign,LastExecutedTime,JobStatus)VALUES ('XYZ_ABC_Export',@XYZWorkFlowID ,'Dialer',GETDATE(),'InProgress')I need to get the identity value generated by JobTracker table and assign it to a ‘JobTrackerId’ pipeline parameter. I need to use the same parameter in the stop event jdbc to update JobTracker the table.I tried adding: ${JobTrackerId} = SELECT @@IdentityI'm pretty sure this is the wrong approach. Can anyone guide me through the right one?

110
mithunvs47
Fan
C
Fan
chrisFan
asked in Community Articles and Got a Question?
Using expression language to get the length of a list

Docs and at least one community thread have suggested that the following expression should work in a stream selector conditional:${record:exists('/ids') && length(record:value('/ids')) > 0}This returns:ELException: No function is mapped to the name "length".The field in question is a list of integers. What am I not understanding?Thanks

90
C
Fan
S
Fan
SALMA AHMEDFan
asked in Community Articles and Got a Question?
Achieving Pagination 

I have resource URL as “https://api.simpleng.com.au/v2/ActionSheets/SummaryForSite/3?data=bma&changesAfter=2024-11-01T00:00:00&pageNumber=1” And response as :{    "page": 1,    "ofPages": 3,    "content": [ ] }Where I want to get all records in Page 1,2 and 3.Despite using pagination I am not able to get the records from all the pages and get only from 1st page.I am using Pagination with PageNumber and using ${StartAt} variable.Refer below Screenshots  

170
S
Fan
hamilton_ferrari
Fan
hamilton_ferrariFan
asked in StreamSets Academy
Can we have multiple columns in JDBC Query Concumer Offset Column ?

While i am working with a table that had no Primary Key and the total number of records were also huge, so i thought of a basic SQL concept that “ To Uniquely Identify a row some times we use Multiple columns “.To use the same in incremental mode in JDBC Query Consumer, I thought of using multiple columns as offset, but i have gone nowhere.Can anybody have any thoughts on this.Thank you.

100
hamilton_ferrari
Fan
manaso
Fan
manasoFan
asked in Community Articles and Got a Question?
streamsets error crear pipeline

Hello everybody. I'm doing an exercise on streamsets and when creating a pipeline it throws this error.VALIDATION_0096 - This pipeline was created in version 6.1.0 and is not compatible with current version 6.0.0.How to fix it? I'm using the 30-day trial version of streamsets.Thank you very much.

191
Sanjeev
StreamSets Employee
J
Fan
jcuevaFan
asked in Community Articles and Got a Question?
Issue Connecting to Redis on IBM Cloud Using Certificate Authority

Hello,I am trying to establish a connection from StreamSets to a Redis database on IBM Cloud. This instance requires the use of a Certificate Authority (CA) for authentication and secure communication, but I haven't found a way to properly configure it in StreamSets.I have attempted the following without success: Configuring a Truststore: I created a Truststore and imported the CA certificate using keytool. I updated the StreamSets configuration (sdc.properties), including the https.truststore.path property, to reference the Truststore. Passing parameters in the connection URI: I tried different URI formats, including adding the certificate to the connection string, but the connection still fails. I would appreciate any guidance on the correct way to configure StreamSets to trust the Redis certificate and establish a secure connection.

110
J
Fan
K
Fan
kamakhya05Fan
asked in Community Articles and Got a Question?
CDC operation with MongoDB oplog origin

Dear Community,I hope this message finds you well. I am reaching out to seek your help on an error getting in validation of Mongodb oplog origin stage. Your support would be greatly appreciated.Validation Error: [ MONGODB_04 - Collection isn't tailable because 'oplog.rs' is not a capped collection. : configBean.mongoConfig.collection ]Other detail/info: > Stage Library: MongoDB Java driver 3.12.0 > I tried with another capped collection 'oplog.test' created in development database other than local. But getting the same error. > I checked the older post and found same error is mentioned, however solution is not there.Thank you in advance for your time and assistance.Best regards,Kamakhya

50
K
Fan
venkataraothanga
Fan
venkataraothangaFan
asked in Share your Best Practices
StreamSets Data Collector 5.9.1 vs StreamSets Data Collector 6.1.0

We are in the process of upgrading from StreamSets Data Collector version 5.9.1 to 6.1.0 and StreamSets Transformer version 5.7.0 to 6.0.0. What are the major changes and features in the latest versions, and what has been deprecated? Which version control hub is supported? Additionally, what considerations should we keep in mind for the upgrade?  

220
venkataraothanga
Fan
F
Fan
fdebordeFan
asked in Community Articles and Got a Question?
Data rule : trigger alert when a field (STRING) is not empty

Hi everyone, I would like to trigger an alert when a field “text” is not empty (from Directory stage with Data Format as Text). Nevertheless, even if text is not empty, the alert is not triggered.I don’t understand what is wrong in the data rule. Any idea ?I have read the documentation here : https://docs.streamsets.com/platform-datacollector/latest/datacollector/UserGuide/Tutorial/BasicTutorial.html#task_ets_dkj_ps.Thanks, Flo

150
F
Fan
F
Fan
fdebordeFan
asked in Community Articles and Got a Question?
Metric rule : trigger alert when processing time through entire pipeline > 2000 ms

Hi everyone, I would like to trigger alert when processing time through entire pipeline > 2000 ms.I have tried with this, but it is not working well.I have tried several other options but I can’t make it work. Any idea ? Thanks, Flo

180
F
Fan
himanshu1234567
Discovered Fame
himanshu1234567Discovered Fame
asked in Community Articles and Got a Question?
How to read files from s3 bucket using groovy?

Can someone provide working code of groovy to read files from s3 bucket?

3018
Nitika
Fan
T
Fan
tomFan
asked in Community Articles and Got a Question?
How to Install Elasticsearch 8.x Stage Library in StreamSets 3.22.3?

Is the Elasticsearch 8.x stage library officially supported in StreamSets 3.22.3? If not, is there a workaround or manual method to install and use Elasticsearch 8.x with this version of StreamSets? Should I consider upgrading to a newer version of StreamSets, or is there another solution?I appreciate any guidance or insights from the community. Thank you!

120
T
Fan
J
Fan
JMABFan
asked in Community Articles and Got a Question?
Start and Stop Event Connection via Parameter

Hi, I am currently developing a pipeline with Start and Stop Events using JDBC query. We are trying to make it more dynamic by having the connections parameterized, however, Start and Stop Events configuration won’t allow to use a parameterized connection.  Even if I click the “Use parameter” button, it does not work. Can anyone please help me on this?

150
J
Fan
R
Fan
rajasriramFan
asked in Community Articles and Got a Question?
Read Entire JSON data as wholefile format in ADLS and write to VARIANT Field in Snowflake

I have a use case to read a JSON file from ADLS location and write the entire content of the JSON file to a VARIANT field in Snowflake. I have used ADLS origin stage , Jython Processor and Snowflake Destination Stage. In the ADLS stage, I have the data format as “WholeFile” to get /fileRef.I used the Jython processor to read /FileRef and convert it as String before loading this field to Snowflake.I am getting error in Jython Processor Stage Script sent record to error: 'com.streamsets.pipeline.stage.util.scripting.Scrip' object has no attribute 'createInputStreamfileRef : FILE_REF[object Object]How do I read entire JSON file and load the contents to a VARIANT field in SnowflakeSample JSON:{ "MbrDetails" : { "id" : "100001", "ProgramName" : "test"},"NotificationInformation" : { "ProgramStatusDate" : "2024-11-13", "Category" : "Pilot Program, "DateTime" : "2024-11-13", "ProgramStatus" : "Eligible"} } Expected Outcome in Snowflake Table:full_json file_name { "MbrDetails" :  { "

70
R
Fan
dannydineen96
Fan
dannydineen96Fan
asked in Community Articles and Got a Question?
Creating a scheduler via Control Hub API

I want to create a scheduler via the following API request: Creates a new Scheduled Task: PUT {SCH_URL}/scheduler/rest/v2/jobs/{pageId}I have 2 questions: 1) Do I use {pageId} = SchedulerJobListPage or SchedulerJobManagePage, and why?Secondly, I’m passing a request body as per the body suggested on the RESTful API Scheduler page but I’m receiving the error: “ERROR : READ_ONLY_ENFORCER_000 - Page ‘com.streamsets.apps.scheduler.rest.SchedulerJobManagePage’ readOnly values have been modified. 2) Is this error due to the readOnlySignature i have generated or with one of the attributes in the request body?  

333
Sanjeev
StreamSets Employee
junsuk
Fan
junsukFan
asked in Community Articles and Got a Question?
Cannot configure a StreamSets pipeline with OpenSearch version 2.11 or above

Hello. I want to store data in OpenSearch internally, but I keep getting the following error. ELASTICSEARCH_17 - Could not index '1000' records: POST https://testpate.net/_bulk: HTTP/1.1 400 Bad Request {"error":{"root_cause":[{"type":"illegal_argument_exception","reason":"Action/metadata line [1] contains an unknown parameter [_type]"}],"type":"illegal_argument_exception","reason":"Action/metadata line [1] contains an unknown parameter [_type]"},"status":400}I tried configuring the pipeline with the latest version of StreamSets, but I'm still encountering the same issue. I've tested both version 7.9.0 and 8.1.0, and I used _doc for all mapping values. Does this mean I can't use OpenSearch?

170
junsuk
Fan

Badge winners

Show all badges
Terms & Conditions