StreamSets Platform

Hello,  I am creating the most simplest pipeline:  a query (JDBC Query Consumer) whose output is written to files - Local FS.  I have broken the files to 200k records each but this job never stops running and keeps writing file after file of what I assume is duplicate data. I cannot for the life of me figure out what I did wrong.  Any insight would be greatly appreciated!I have created complex pipelines in the past, so this is strange.  

The HTTP Client in my pipeline is not processing all of the input records that it gets. For eg: Input to HTTP client is 1430 records, but the output records processed in the same client are 1360 records only, with 0 error records. Not sure if I am missing any configuration to be added so that I can balance the I/O records, and send the error records to the error stage.

Hello There I am trying to solve a very specific usecase here. I am trying to query a DB using an API and this API is using multiple query parameters. one of the query parameters is going to be ids which are more than 100 in count. The catch is that I can’t pass all 100 ids as an array to the API call because API is not designed to accept array for that parameter. It is going to be kind of looping over those 100 ids one by one and then calling the API with new id as the parameter value in each iteration.Also, these IDs needs to be fetched from a snowflake table and then passed as the parameter to the API call. So I am thinking of having some snowflake or JDBC query consumer as as an origin. And these IDs would increase over the period of time so want to make it as dynamic as possible but that is not priority for now.  Having multiple jobs to solve this would lead to 100+ jobs and that would keep increasing which is not a good practice at all. Could someone please suggest the best possi

Hi,In my pipeline, I am having a stream selector stage. I want to parameterize it and use the following expression for the condition:${record:value('/rating_text') == '${pipeline_rating_text}'}Here, pipeline_rating_text is my parameter that I have defined for my pipeline. The problem is that when I run the pipeline it does not work. If I use this expression ${record:value('/rating_text') == 'Excellent'}everything will be fine. Can somebody help me, please?

Hi,I have a JDBC connection for a database that is located in a docker container in my local machine. This connection works perfectly when I make a data collector pipeline.I installed a transformer engine in a docker container in my local machine (I have installed the external JDBC libraries). Then, I made a very simple pipeline to read from my database using this JDBC connection. I constantly get this error message that “[JDBC Table 1] Cannot connect to specified database: Communications link failure The last packet sent successfully to the server was 0 milliseconds ago. The driver has not received any packets from the server. (JDBC_00)”.Can anyone help me with solving the problem?

Dears,I have configured a local docker image for the Streamsets and kafka images to try a simple Kafka connection, but I´m receiving the following:Http failure response for https://na01.hub.streamsets.com/tunneling/rest/660c2f92-c396-4322-a9ea-cd73758897a1/rest/v1/pipeline/dynamicPreview?TUNNELING_INSTANCE_ID=tunneling-1: 500 OK The kafka image names are:bash-3.2$ docker-compose psNAME                IMAGE                    COMMAND                  SERVICE             CREATED             STATUS              PORTSkafka               wurstmeister/kafka       "start-kafka.sh"         kafka               27 minutes ago      Up 27 minutes       0.0.0.0:9092->9092/tcpzookeeper           wurstmeister/zookeeper   "/bin/sh -c '/usr/sb…"   zookeeper           27 minutes ago      Up 27 minutes       22/tcp, 2888/tcp, 3888/tcp, 0.0.0.0:2181->2181/tcp And I try the Test connection kafka:9092 or localhost:9092 and receive the Http failure response for https://na01.hub.streamsets.com/tunneling

I have passed a parameter to my pipeline which has a default value when I start the pipeline. I need to change it's value in from one of processors. Is it possible to change the runtime parameter? If yes, can you please provide the steps to do that.

I’m getting this error when I’m trying to use Snowflake uploader. I manually created the stage and I know I can put files there. I’m on 5.8 data collector.  How do I over come this error? 

HiI am trying to establishing a connection to jdbc:mysql://mysqldb:3306/zomato   but it is not working.I tried 2 differents approach:at first, I did not want to use Strigo so I was not limited to the 8 hours. So I run my engines in containers within my local machine. The engines are running fine on the Control Hub and I can create and run some pipelines. However, I would like to use the connection cited in one of the lab with mysqldb and zomato. I am not sure how to “install” the zomato db and reviews table locally? I have a container running mysql but from there i dont know what to.         If i try to create the connection using my local engine, i get then the below error.JDBC_00 - Cannot connect to specified database: com.mysql.cj.jdbc.exceptions.CommunicationsException: Communications link failure The last packet sent successfully to the server was 0 milliseconds ago. The driver has not received any packets from the server. I could see a post with the same error message but it does

We are logging pipeline errors /orchestrated task errors in database . Now often, we see error messages like - “for actual error , open the logs “ as runtime errors.Per our solution , this gets logged into the database . We don’t want that ..Instead we want the actual errors to get logged . How can we do that ?

The case is simple,I have a Kafka topic to which the data gets pushed, I use the data to execute a stored procedure using the JDBC query node. This is working fine but I need to get the return value from the SP so I can log if the execution is a success or there was any error in input which is returned by the SP itself. The SP has some complex logic that involves multiple tables so it's not really good I think to eliminate the SP. Since I don't see any output in JDBC query executor how do I redo the pipeline so I can store/process the SP outputs?

I have a simple pipeline from oracle multi table into snowflake destination. This error only happens when the destination table is not present but it does create the table with PK. After pipeline ran and I can truncate the table and re-run the pipeline without any issues.  Any ideas? 

Hello!I'm currently trying out the free (open source version) of SDC. When I created an empty pipeline and added a cron scheduler and JavaScript Evaluator and tried to run it, I encountered the following error:Pipeline Status: STARTING_ERROR: java.lang.NoClassDefFoundError: Could not initialize class com.streamsets.pipeline.stage.processor.javascript.Java8JavaScriptObjectFactoryI was under the impression that using the JavaScript Evaluator did not require any additional package installations. Is there an additional step that needs to be taken?I need some help. Thank you.

Hello,I'm currently using the HTTP Client Processor in StreamSets Data Collector and I've encountered an issue with the retry mechanism. Despite configuring the processor to retry on receiving certain HTTP status codes, it doesn't seem to be doing so.Specifically, I've set the processor to retry immediately when it receives a 409 status code, with a maximum of 2 retries. However, when the processor receives a 409 status code, it doesn't retry the request and instead gives the following error: "HTTP_101: Applying passthrough and error policy on status configuration".I've checked the pipeline configuration. I'm not sure why the processor isn't retrying as expected.Has anyone else encountered this issue? Any insights or suggestions would be greatly appreciated.Thank you in advance for your help.

Hi Team While installing DC Tarball engine using install script from Dataops platform, it asks for download and install directory at run time.Is there a way we could avoid passing it during execution and pass it in install script or use current directory? 

Hello,I'm currently working on a simple pipeline to ingest kafka messages inside a log file.I'm trying to consume all the data from the beginning of a topic but i'm only getting newer data added to this topic.Once consumed, the previous topic messages are not accessible anymore.I've already test all the different "Auto Offset Reset" properties.Same for simple and multi topic consumers.In the official documentation docs.streamsets.com :auto.commit.interval.ms bootstrap.servers enable.auto.commit group.id max.poll.recordsIf I understand correctly all those parameters are locked so I can't disable the offset management and process all the data from the beginning of a topic.Is there an additionnal Kafka configuration property to use or do I need to configure the topic directly via kafka CLI ??StreamSets Data Collector version : 3.14.0Kafka Consumer version : 2.0.0Regards.

Hi Team,We have some pipelines, which has some configuration like below, How can I get the exact plain test value for these parameter property value? Also how and where can I set these kind of values?   

I read from an SFTP server and when writing to S3 my files get renamed. How can I retain the original file name?

Hello Team,I need to update one job config which is “globalMaxRetries” using rest call.I refer Restful section in Control Hub but that Rest call for “updateJob” requires many other configs also needs to mention in Request Body. could you please assist me with this ?

Hi,We are using Microsoft SQL CDC Client to extract changes on one of our database. Now we noticed that the target (a Snowflake destination) is complaining about unknown operations.So added this following step to the data flow:       if (records[i].attributes['sdc.operation.type'] == 5)         continue;         sdc.output.write(records[i]);Is it normal that we have to filter out operation type 5 manually? So that the pipeline works.Is there an example pipeline (SQL CDC → Snowflake) so that we could apply best practicesThanksSebastian

I have developed a pipeline that utilizes an Http Server as the origin. In the configuration of the origin, I have designated port 8000 for the purpose of writing data to the pipeline. However, the issue lies in the fact that the pipeline does not release this port. I aim to utilize the same port for multiple pipelines, distinguishing them by their respective Application-Id. Upon executing the initial pipeline, the port remains occupied and fails to release. Upon attempting to execute a second pipeline using the same origin, I encountered an error message stating "Port is not free."If anyone got solution, Please Help me out with this.

Hello community, I am vuilding a pipeline that receives http requests providing a JSON payload in the body.I want to validate the payload against a json schema. Now I have come accross this article which is interesting https://streamsets.com/blog/json-validator/. I keep getting this error : JSON_VAL_02 - The JSON object supplied is invalid: org.json.JSONException: Expected a ':' after a key at 6 [character 7 line 1] When I check the record as json string checkbox, it raises the exception … com.streamsets.pipeline.api.impl.TypeSupportConversionException: Cannot convert Map to com.streamsets.pipeline.api.impl.StringTypeSupport@374848aAlso with that processor component I am restricted as I would like to fetch the json schema from a repository.I have tried using the javascript evaluator as well and also there I am running into trouble with library versions I believe …Below the error I am getting …Caused by: java.io.IOException: resource /draftv4/schema not foundUsing libraries …json-schem

Hello all,I am trying to run a pipeline with an HTTP server origin to receive messages and then push them to a kafka topic. When I start my docker container, the port is not accessible on my local laptop when I don’t enable port-forwarding for the port the origin is listening on. But when I enable port-forwarding then I get the notification : “The engine cannot be reached. Check that the engine is running and can communicate with Control Hub.” Can anyone help me out on this?Many thanks!Hamid K.

Hi Everyone.We are working in a solution which integrates GCP Secret Manager and Streamsets.The client stores a secret in JSON format, and wants to extract a field with his data. Is similar to this format:   How can I extract the dbuser field??I now there are custom functions, but don’t know if I can use the EL function from Streamsets to obtain the JSON info.Can I use the ${credential:get("gcp", "group@org", "dbuser?latest") as an argument? so the function returns only the dbuser value. Thanks! 

Hi,Use case.We are load data from Teradata to Snowflake.as we not doing any transformation just extract the Data and loading in Snowflake.which engine will be the best for this use case.1.Data collector​​​​​​​2.Transformer for spark.Note: we have data in millions  Best regards,Hrishikesh