StreamSets

# Download StreamSets Data Collector
wget https://archives.streamsets.com/datacollector/5.8.0/tarball/streamsets-datacollector-core-5.8.0.tgz

# Extrair e configurar
tar -xzf streamsets-datacollector-core-5.8.0.tgz
cd streamsets-datacollector-5.8.0

# Configurar Java
export JAVA_HOME=/usr/lib/jvm/java-8-openjdk

# Iniciar Data Collector
bin/streamsets dc

# Acessar interface web
# http://localhost:18630

# HTTP configuration
http.port=18630
http.authentication=form
http.realm.file.permission.check=true

# HTTPS configuration (opcional)
https.port=18631
https.keystore.path=keystore.jks
https.keystore.password=password

# Pipeline configuration
pipeline.max.runners.count=50
runner.thread.pool.size=50

# Stage libraries
stage.libs.dir=${SDC_DIST}/streamsets-libs

# Data directories
sdc.data.dir=${SDC_DATA}/data
sdc.log.dir=${SDC_DATA}/log
sdc.resources.dir=${SDC_DATA}/resources

# Security
kerberos.client.enabled=false
ldap.authentication.enabled=false

# Monitoring
observer.queue.size=100
observer.sampled.records.cache.size=10000

{
  "pipelineId": "kafka-to-hdfs-pipeline",
  "title": "Kafka to HDFS Pipeline",
  "description": "Stream data from Kafka to HDFS with transformations",
  "stages": [
    {
      "instanceName": "KafkaConsumer_01",
      "library": "streamsets-datacollector-apache-kafka_2_6-lib",
      "stageName": "com_streamsets_pipeline_stage_origin_kafka_KafkaDSource",
      "configuration": [
        {
          "name": "conf.brokerURI",
          "value": "localhost:9092"
        },
        {
          "name": "conf.topic",
          "value": "user-events"
        },
        {
          "name": "conf.consumerGroup",
          "value": "streamsets-consumer"
        }
      ]
    },
    {
      "instanceName": "JSONParser_01",
      "library": "streamsets-datacollector-basic-lib",
      "stageName": "com_streamsets_pipeline_stage_processor_jsonparser_JsonParserDProcessor",
      "configuration": [
        {
          "name": "fieldPathToParse",
          "value": "/text"
        },
        {
          "name": "parsedFieldPath",
          "value": "/parsed"
        }
      ]
    },
    {
      "instanceName": "HadoopFS_01",
      "library": "streamsets-datacollector-hdp_3_1-lib",
      "stageName": "com_streamsets_pipeline_stage_destination_hdfs_HdfsDTarget",
      "configuration": [
        {
          "name": "hdfsUri",
          "value": "hdfs://localhost:9000"
        },
        {
          "name": "hdfsUser",
          "value": "hdfs"
        },
        {
          "name": "dirPathTemplate",
          "value": "/data/events/${YYYY()}-${MM()}-${DD()}"
        },
        {
          "name": "fileNamePrefix",
          "value": "events"
        }
      ]
    }
  ]
}

@StageDef(
    version = 1,
    label = "Custom Data Enricher",
    description = "Enriches records with external data",
    icon = "enricher.png",
    onlineHelpRefUrl = ""
)
@ConfigGroups(Groups.class)
@GenerateResourceBundle
public class CustomEnricherProcessor extends SingleLaneProcessor {

    @ConfigDef(
        required = true,
        type = ConfigDef.Type.STRING,
        defaultValue = "http://api.example.com",
        label = "API Endpoint",
        description = "External API endpoint for enrichment",
        displayPosition = 10,
        group = "ENRICHMENT"
    )
    public String apiEndpoint;

    @ConfigDef(
        required = true,
        type = ConfigDef.Type.STRING,
        defaultValue = "/userId",
        label = "Key Field",
        description = "Field to use as lookup key",
        displayPosition = 20,
        group = "ENRICHMENT"
    )
    public String keyField;

    private WebTarget webTarget;

    @Override
    protected List init() {
        List issues = super.init();
        
        try {
            Client client = ClientBuilder.newClient();
            webTarget = client.target(apiEndpoint);
        } catch (Exception e) {
            issues.add(getContext().createConfigIssue(
                Groups.ENRICHMENT.name(),
                "apiEndpoint",
                Errors.CUSTOM_01,
                e.getMessage()
            ));
        }
        
        return issues;
    }

    @Override
    protected void process(Record record, SingleLaneBatchMaker batchMaker) 
            throws StageException {
        
        try {
            // Extract key value
            String keyValue = record.get(keyField).getValueAsString();
            
            // Call external API
            Response response = webTarget
                .path("lookup")
                .queryParam("key", keyValue)
                .request(MediaType.APPLICATION_JSON)
                .get();
            
            if (response.getStatus() == 200) {
                String enrichmentData = response.readEntity(String.class);
                
                // Add enrichment data to record
                record.set("/enrichment", Field.create(enrichmentData));
                
                batchMaker.addRecord(record);
            } else {
                // Handle error
                getContext().reportError(Errors.CUSTOM_02, keyValue);
            }
            
        } catch (Exception e) {
            throw new StageException(Errors.CUSTOM_03, e.getMessage(), e);
        }
    }
}

# Dockerfile
FROM streamsets/datacollector:5.8.0

# Install additional stage libraries
RUN "${SDC_DIST}/bin/streamsets" stagelibs -install=streamsets-datacollector-apache-kafka_2_6-lib
RUN "${SDC_DIST}/bin/streamsets" stagelibs -install=streamsets-datacollector-hdp_3_1-lib

# Copy custom configurations
COPY sdc.properties "${SDC_CONF}/sdc.properties"
COPY pipeline.json "${SDC_DATA}/pipelines/"

# Expose ports
EXPOSE 18630

# Docker Compose
version: '3.8'
services:
  streamsets:
    build: .
    ports:
      - "18630:18630"
    environment:
      - SDC_JAVA_OPTS=-Xmx2g -Xms2g
    volumes:
      - streamsets_data:/data
      - streamsets_logs:/logs
    networks:
      - data_network

  kafka:
    image: confluentinc/cp-kafka:latest
    environment:
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:9092
    networks:
      - data_network

volumes:
  streamsets_data:
  streamsets_logs:

networks:
  data_network:
    driver: bridge