Streams Via Config Files

When running Redpanda Connect in streams mode it’s possible to create streams with their own static configurations, simply list one or more files after the streams subcommand:

rpk connect streams ./foo.yaml ./configs/*.yaml

Resources

A stream configuration should only include the base stream component fields (input, buffer, pipeline, output), and therefore should NOT include any resources. Instead, define resources separately and import them using the -r/--resources flag:

rpk connect streams -r "./resources/prod/*.yaml" ./stream_configs/*.yaml

Walkthrough

Make a directory of stream configs:

$ mkdir ./streams

$ cat > ./streams/foo.yaml <<EOF
input:
  http_server: {}
pipeline:
  threads: 4
  processors:
    - mapping: 'root = {"id": this.user.id, "content": this.body.content}'
output:
  http_server: {}
EOF

$ cat > ./streams/bar.yaml <<EOF
input:
  kafka:
    addresses:
      - localhost:9092
    topics:
      - my_topic
pipeline:
  threads: 1
  processors:
    - mapping: 'root = this.uppercase()'
output:
  elasticsearch:
    urls:
    - http://localhost:9200
EOF

Run Redpanda Connect in streams mode, pointing to our directory of streams:

rpk connect streams ./streams/*.yaml

On a separate terminal you can query the set of streams loaded:

curl http://localhost:4195/streams | jq '.'
{
  "bar": {
    "active": true,
    "uptime": 19.381001424,
    "uptime_str": "19.381001552s"
  },
  "foo": {
    "active": true,
    "uptime": 19.380582951,
    "uptime_str": "19.380583306s"
  }
}

You can also query a specific stream to see the loaded configuration:

curl http://localhost:4195/streams/foo | jq '.'
{
  "active": true,
  "uptime": 69.334717193,
  "uptime_str": "1m9.334717193s",
  "config": {
    "input": {
      "http_server": {
        "address": "",
        "cert_file": "",
        "key_file": "",
        "path": "/post",
        "timeout": "5s"
      }
    },
    "buffer": {
      "memory": {
        "limit": 10000000
      }
    },
    "pipeline": {
      "processors": [
        {
          "mapping": "root = {\"id\": this.user.id, \"content\": this.body.content}",
        }
      ],
      "threads": 4
    },
    "output": {
      "http_server": {
        "address": "",
        "cert_file": "",
        "key_file": "",
        "path": "/get",
        "stream_path": "/get/stream",
        "timeout": "5s"
      }
    }
  }
}

You can then send data to the stream via its namespaced URL:

curl http://localhost:4195/foo/post -d '{"user":{"id":"foo"},"body":{"content":"bar"}}'

There are other endpoints in the REST API for creating, updating and deleting streams.