StreamSets Platform

Hello everyone, I have a csv file and I want to send data to a REST API. My final data looks like this: I will give you an example of how I want my final result to be. For 3 records in my csv file, every record will look similar with the one above. For 1 record in my csv file, I want to sent firstly a json with serial, interface and userData_dep fields, and then another json with serial, interface and userData_lan fields. To do that I used 2 http client processors, one for the dep and one for the lan. But I saw that the jsons weren’t sent as I want. My pipeline send 3 json with _dep field, and then another 3 json with _lan field. I want to send 1 json with dep field, then 1 json with lan field, then the next json with the dep field, and the next json with the lan field and so on. I want to send them sequentially. Could you help me with some ideas? Thank you!

We understand the Oracle CDC client for CDC from Oracle using SDC. But does SDC support Golden Gate as CDC origin please?

I have a data collector pipeline that runs on data collector version 3.21.x This pipeline reads data using a JDBC Multi-table Consumer Origin and writes to a Snowflake DB/table. The pipeline used to run successfully. But in the past two days we encounter the following error, "errorMessage": "SNOWFLAKE_59 - Could not get a Stage File Writer instance to write the records: 'SNOWFLAKE_65 - File upload to Snowflake failed: net.snowflake.client.jdbc.SnowflakeSQLLoggedException: S3 operation failed: Operation=listObjects, Error type=Client, Error code=AccessDenied, Error message=Access Denied (Service: Amazon S3; Status Code: 403; Error Code: AccessDenied; Request ID: DK041FE70QRW4B85; S3 Extended Request ID: DCSX2ikBaa3k/Vwcsc5+Cbb+pH+7KkLITtJYr+e3ea7cHlOVvVeDG9xhpZtQnbc/6FrAsef4C/w=), Request id=DK041FE70QRW4B85, Extended request id=DCSX2ikBaa3k/Vwcsc5+Cbb+pH+7KkLITtJYr+e3ea7cHlOVvVeDG9xhpZtQnbc/6FrAsef4C/w='", Note: I am using the Snowflake Internal Stage NOT AWS S3 and hence unsure what p

In order to send the file from source system to SFTP/MFT servers , we need to select the data format as “Whole file” in the origin because the sftp processor only accept the data format as whole file.

Preface Fetching online and offline data across different channels to understand your end-users behavior is a primary topic for any enterprise-level brand. Customer 360 view of an end customer gives enormous insight to marketing leaders who can then prepare their strategy to engage and entice these users. A full-stack CDP (Customer Data Platform) helps meet the needs of marketing leaders and analytic folks. A stable, reliable, and scalable data processing stack is the de facto need for any enterprise CDP solution, and to set up that, data ingestion and extraction architecture need to be solid. StreamSets Data Collector (SDC) has become the preferred tool for many data engineers and data stewards to achieve the ETL and ELT-related needs in this area. In the traditional approach, the ETL engineers normally have definitive requirements in a place where they know the origin, transformation rules, and destination. Based on the requirements, engineers develop their pipelines and deploy them

Hello, I am using a Field Type Converter to convert string to datetime with date format (yyyy-MM-dd'T'HH:mm:ss.SSS'Z' - 2021-08-18T08:20:24.136000Z). However, the same field has dates with format 2021-08-18T08:20:24.00Z, which it gives me error CONVERTER_00 - Failed to convert field '/createdate' of type 'STRING' with value '2021-08-19T14:01:32Z' to type 'DATETIME' Any recommendations on how to handle two different date formats for the same field?

Overview Since November 2014 we have been implementing a Data Lake platform to collect, process, filter, enrich, store, analyse and visualize logs and business events using Big Data and streaming technologies. The main goal is to detect important events, generate statistics and produce alerts to solve problems in a proactive way. Our first pilot on a new customer is usually related to application logs but once this platform is up and running it naturally becomes the centralized repository not only for logs but also for any type of sources and data formats. With dashboards on display in big screens other areas and departments start requesting dashboards of their own that require new data sources to be connected. To adapt to different use cases and new requirements we needed a more modular, flexible and extensible architecture. As the Data Lake gained prominence within our firsts customers and other project opportunities materialized, challenges and constraints emerged that led us to ret

I get this error, following the steps in the lecture “DataOps Platform Fundamentals”, chapter “Build a Pipeline”, lab “Lab: Build a Pipeline”. When clicking on the error, it takes me to the Stream Selector module, Configuration, General tab. There is nothing highlighted in that section and all seems in working order to me. Could you please advise?

Hello I tried to extract and transform data from an CSV file and to obtain as an output a JSON object. I used the JSON generator and in the preview area things looked good. I will attach here the output. Output from DC I tried to sent this object to a REST API but when I looked into the logs, I realized that the format of the input was not the same as I expected. Some backslashes are added before every field and every value. Input from logs Could you please help me with some ideas or maybe a solution to my problem? Thank you, Diana

Dear Experts, I have a simple SFTP data pipeline to take text files from an SFTP cloud server and push these files to a local file system. However, I am required to get these files scanned by an anti-virus via ICAP call "icap://${SERVER_IP}:${SERVER_PORT}/virusScan" before writing to the target. Can anyone guide me on how to get this done in Streamsets? FYI: some other tools like Informatica have a built-in option for this. Please refer this link: https://knowledge.informatica.com/s/article/613287?language=en_US&type=external

Hi I am on Data Collector StreamSets Data Collector 3.20.0 and received a new error on a production pipeline Pipeline Status: RUN_ERROR: HADOOPFS_13 - Error while writing to HDFS: org.apache.hadoop.fs.FileAlreadyExistsException: Operation failed: "This endpoint does not support BlobStorageEvents or SoftDelete. Please disable these account features if you would like to use this endpoint.", 409, HEAD, https://gcdatalake2.dfs.core.windows.net/data-warehouse//?upn=false&action=getAccessControl&timeout=90 This pipeline was working, and now any of my pipelines that require access to any Azure Blobstore is StackTrace is com.streamsets.pipeline.api.StageException: HADOOPFS_13 - Error while writing to HDFS: org.apache.hadoop.fs.FileAlreadyExistsException: Operation failed: "This endpoint does not support BlobStorageEvents or SoftDelete. Please disable these account features if you would like to use this endpoint.", 409, HEAD, https://gcdatalake2.dfs.core.windows.net/data-warehouse//?upn=false&a

Greetings, I’d like to use the Subscription option to send notifications to a Slack channel. Most of the fields make sense but I am stuck on Content Type and Payload. I searched in Documents but didn’t find anything that covered Subscriptions. Did I miss it? -David

I have a use-case, my org. is using Data Collector 3.21.x for ingesting non-sensitive data to a S3 bucket. We currently have a requirement to bring sensitive data through data collector into a S3 bucket (specifically for sensitive data). My questions, I do not want all users who have access to StreamSets Data Collector to view the pipeline that brings the sensitive data. Can I do that with the open source version (3.21.x) of Data Collector? If yes, please advise How? Would it be possible to restrict users to view or NOT view a set of pipelines in StreamSets Data Collector (open source version). Example: I want ONLY members of the Finance team to view Finance pipelines in Data Collector. Note: I know, I can spin up a separate instance of Data Collector just to handle sensitive data but I do not want to go down that path. Thanks, Srini

I'm running SDC in a Docker container and I am trying to invoke a HTTP call with HTTP Client using a restricted header like below: curl -X GET -H "Host: my-host" http://my-service/my-endpoint However, by default the header is ignored with the following warning: HttpUrlConnector - Attempt to send restricted header(s) while the [sun.net.http.allowRestrictedHeaders] system property not set. Header(s) will possibly be ignored Where do I set this property? Adding it to SDC config did not help.

Is there a more sophisticated way of removing backslash within a string? I've tried string.replace(/\\/g, ""), but that did not help.

When writing to Amazon S3 destination why are these “WARN” messages showing up in the logs? Does this have any impact on the running pipeline? No content length specified for stream data. Stream contents will be buffered in memory and could result in out of memory errors.

When writing to Kafka Producer destination it doesn’t always handle timeout exceptions (as shown below) and the pipeline does not honor On Record Error » Send to Error setting on the Kafka Producer destination. How can this be resolved? Caused by: java.util.concurrent.ExecutionException: org.apache.kafka.common.errors.TimeoutException: Failed to update metadata after 60000 ms. at org.apache.kafka.clients.producer.KafkaProducer$FutureFailure.<init>(KafkaProducer.java:1186) at org.apache.kafka.clients.producer.KafkaProducer.doSend(KafkaProducer.java:880) at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:803) at org.apache.kafka.clients.producer.KafkaProducer.send(KafkaProducer.java:690) at com.streamsets.pipeline.kafka.impl.BaseKafkaProducer09.enqueueMessage(BaseKafkaProducer09.java:64) at com.streamsets.pipeline.stage.destination.kafka.KafkaTarget.writeOneMessagePerRecord(KafkaTarget.java:242) ... 30 more Caused by: org.apache.kafka.common.errors.TimeoutExceptio

Team, We are processing data from the Hadoop Avro file to SQL server DB. Whenever the SS process is running, SQL DB performance is degrading due to Mapper execution. We checked with DBA and they suggested decreasing the Insert connections. We need to decrease the mappers (stable) for this data load. Is there any way to limit the mapper creation/insert operation? Appreciate your help!

Hi Team, I am trying to read the files from AWS S3 using the Amazon S3 origin. It does read the file at first level but doesn’t read the file recursively. In the below image, for bucket “s3a://sotero-transformer/input/”, it reads the files userdata1.parquet and userdata2.parquet but doesn’t read the files under “dirone”. What should be the configuration to read all the files under S3 bucket recursively?

We want to create an integration test suite and run it in a self-contained world of headless containers The high level goal is docker-compose to start all containers streamsets kafka configure streamsets send data to kafka topic StreamSets processes the data in the topic and send its to other topics Application under tests, processes the data (fails / passes) - the test tear down Question: How do I configure streamset pipeline without using the UI?

After going through the configuration for Data bricks Delta lake destination ,I’m trying to partitioning table based on partition column but encountered with an "DELTA_LAKE_41" error message.

Trying to read XML in the following format with “:” is throwing Can't parse XML element names containing colon ':' error. <sh:root> <sh:book> </sh:book> <sh:genre> </sh:genre> <sh:id> </sh:id> <sh:book> </sh:book> <sh:genre> </sh:genre> <sh:id> </sh:id> <sh:book> </sh:book> <sh:genre> </sh:genre> <sh:id> </sh:id></sh:root> What’s the best way to read such XML? (Note that changing “:” to “_” in the XML works.)

I'm trying to enable Kerberos for my SDC RPM installation, but when I start the SDC I get following exception: java.lang.RuntimeException: Could not get Kerberos credentials: javax.security.auth.login.LoginEx Caused by: javax.security.auth.login.LoginException: Unable to obtain password from user at com.sun.security.auth.module.Krb5LoginModule.promptForPass(Krb5LoginModule.java:897) at com.sun.security.auth.module.Krb5LoginModule.attemptAuthentication(Krb5LoginModule.java:760) at com.sun.security.auth.module.Krb5LoginModule.login(Krb5LoginModule.java:617) at sun.reflect.NativeMethodAccessorImpl.invoke0(Native Method) at sun.reflect.NativeMethodAccessorImpl.invoke(NativeMethodAccessorImpl.java:62) at sun.reflect.DelegatingMethodAccessorImpl.invoke(DelegatingMethodAccessorImpl.java:43) at java.lang.reflect.Method.invoke(Method.java:498) How do I move forward?

I have a StreamSets Data Collector running in Docker and when I run a pipeline with Kafka Consumer I am seeing these error messages: The configuration = was supplied but isn't a known configThe configuration schema.registry.url = was supplied but isn't a known config How do I get past this error?

I have a few JDBC-based stages in my pipeline (Origin, JDBC Lookup, etc.) and when I try to replace the existing JDBC-based origin with another (for example, Oracle CDC with MySQL Binary Log), the validation just on the lookup processors fails with “Failed to get driver instance with multiple JDBC connections” error. Even though I haven’t changed anything on those processors. Here’s the stack trace… java.lang.RuntimeException: Failed to get driver instance for jdbcUrl=jdbc:oracle:thin:@connection_URL at com.zaxxer.hikari.util.DriverDataSource.<init>(DriverDataSource.java:112) at com.zaxxer.hikari.pool.PoolBase.initializeDataSource(PoolBase.java:336) at com.zaxxer.hikari.pool.PoolBase.<init>(PoolBase.java:109) at com.zaxxer.hikari.pool.HikariPool.<init>(HikariPool.java:108) at com.zaxxer.hikari.HikariDataSource.<init>(HikariDataSource.java:81) at com.streamsets.pipeline.lib.jdbc.JdbcUtil.createDataSourceForRead(JdbcUtil.java:875) at com.streamsets.pipeline.stage.processor.jdbclookup.Jdb