Create a curated stream of data for other teams

Select Apache Kafka (or your system of choice). Then, choose Source as the connection type and provide the required connection information such as the list of bootstrap brokers, authentication method, credentials, and any other optional connection parameters.

Depending on the security protocol that you want to use to connect to Apache Kafka, you may need to provide additional information such as your certificates and keys or your SASL username and password.

The Value Format parameter controls how Decodable deserializes data from the Kafka topic. Decodable supports popular serialization formats like JSON, Avro, and Debezium (JSON) schemas. Make sure to select the format that matches your data.

If Decodable encounters data in a format other than what’s configured, it will follow the Parse Error Policy method you select. Selecting Fail causes Decodable to fail the connection and stop processing until the issue is corrected, while Ignore causes the system to skip and discard the malformed record.

By default, Kafka connections operate in exactly once delivery mode in Decodable. For more information about the other Apache Kafka options, see Apache Kafka.

Select Next.

Streams transport records within Decodable. You can select a preexisting stream to which data should be written, or you can create a new stream.

If you’re creating a new stream, give it a name. Names must be unique within an account and can contain letters, numbers, and underscores ([a-z0-9\_]+). This is how you’ll refer to your data in SQL queries and custom pipelines written in Java. Best practices are to give streams a description that helps you and your team remember what data it contains and how it should be used.

Next, you must define the schema of the stream. The stream’s schema must match the schema of the data in the Postgres database. This is how Decodable validates and enables processing of data. The schema can be as simple as having a single text field or can fully describe a complex nested record. All records in a stream must have the same schema. If you already have a schema defined in JSON Schema or Apache Avro format, you can upload these to Decodable.

To learn more about streams, see About Streams for an overview on the types of streams available, About Change Data Capture and Change Streams for information about change streams, and Create a Stream for information about creating streams, including how to specify event time and watermark information as well as how to use computed or metadata fields.

Give your connection a meaningful name and description for future reference.

Selecting Create Connection performs a connectivity test. If Decodable successfully connects to your Postgres table, a connection is created and the Connection Overview page opens.

If Decodable is unable to validate the connection, an error is shown with more information. The most common type of error is a network connectivity error because Decodable is unable to connect to your database because a firewall or security group prevents access. This often results in a timeout while trying to connect. Make sure you’ve allowed connections from Decodable’s IPs.

Once your connection is created, you can start it by selecting Start. You choose the maximum task size and the maximum number of tasks when activating a connector or pipeline. A task is a compute unit that includes CPU, memory, network bandwidth, and local storage. Most connectors and all pipelines support scale-out parallel execution. The number of tasks controls the maximum parallelism of a connection or pipeline. A medium task is sufficient for most connections and should provide between 1000 and 30,000 records per second depending on record size, serialization format, and similar factors.

Activation can take about a minute while Decodable allocates resources. Your connection should display Starting, and eventually transition to Running. If there’s data on the topic, you should see output record counts and bytes metrics begin to increase. Note that metrics are reset upon each activation. Stopping and restarting a connection resets metrics back to zero.

If you don’t see metrics increasing, make sure that the connection has the status Running.

You can follow the lineage trail by selecting Outbound to 1 stream - your stream name to preview data in the source connection’s output stream. This is the stream you selected or created in step 1.2. Stream preview is updated whenever you navigate to a stream, but it can be manually refreshed by selecting Run Preview.

If you don’t see data in your output stream:

Select Pipelines, and then New Pipeline. If you are participating in the public tech preview for custom pipelines, which allows you write pipelines in Java and other JVM-based languages using the full power of the Apache Flink APIs, you can select between SQL and a custom pipeline. For now, select SQL.

Choose your Input Stream. This should be the stream you created or selected in step 1.2. You can always change your mind later while editing the pipeline SQL.

In the SQL editor you can browse all of your existing streams in the Data Catalog panel, edit SQL, and preview the results of your SQL query.

SQL pipelines are always insert into …​ select …​ from …​ queries. The insert into clause specifies the pipeline’s output stream, while the select …​ from …​ specifies the input streams and any transformations. Decodable automatically generates an output stream name for you (for example stream_<digits>), but you’re encouraged to change it to something meaningful to you. Decodable supports standard SQL including joins, windowed and global aggregation, Common Table Expressions (CTE), and even pattern detection with match_recognize.

For example, let’s assume that our data contained order events. We can filter our data by configuring the following pipeline.

If necessary, we can keep only the necessary fields and mask sensitive data.

See the Function Reference to learn more about supported SQL functions. If you’re interested in windowed aggregations, those are supported too. There are also a number of SQL pipeline examples to help get you started with common patterns.

Next, let’s preview our pipeline to make sure the SQL is processing the data as expected. Previews display up to 30 records from the data stream and can be used to see how your pipeline changes the incoming data. Previews run until 30 records have been seen or until 2 minutes have passed, whichever comes first.

Start the preview by selecting Run Preview. You can optionally run the preview from the earliest position in the input streams or the latest position. By default, preview runs from the latest position.

Not seeing preview output? Here are a few things to check:

Once your data looks correct, select Next to continue.

If you’ve specified an output stream that doesn’t exist, Decodable prompts you to create a new stream. The stream’s schema is automatically inferred from the SQL query, including field names and their types. The stream’s name is defined by the SQL query, so it isn’t editable while displayed. As always, you should give your stream a meaningful description.

Once everything looks correct, select Create Stream.

Alternatively, if you specified a stream that already exists as an output stream in your SQL query, the stream’s schema must match the output of the select …​ from …​ portion of the query. You’ll get an error if this isn’t the case.

Finally, just like our connection, give the new pipeline a name and description.

Select Create Pipeline to complete the process.

Like connections, pipelines must be started in order to begin processing data.

Select Start to activate your pipeline. Select the appropriate maximum task size and maximum task count just as we did for our connection. Like preview, select whether the pipeline should start from the earliest or latest position in the input stream.

Pipeline metrics work identically to connection metrics with a few noteworthy exceptions:

Select Output to 1 stream in the lineage bar to navigate to the pipeline’s output stream and confirm that your pipeline is processing and sending data.

Complete the required fields for the connection to Amazon S3. This includes the file path where your bucket is located, the AWS ARN of the IAM role with sufficient permissions to access the S3 bucket, and the region that your S3 bucket is in.

For more information about the Amazon S3 Connector, including how Decodable constructs object key names in Amazon S3, see the Amazon S3 connector documentation.

Select Next to validate the connection parameters.

Select Start to activate the sink connection. Choose the appropriate maximum task size and maximum number of tasks for your data volume.

Go to Access Control and select Roles. Roles define the permissions that the users of a specific group have. As a Decodable administrator, you can assign roles to any group in your account.

Select New Role, and give your role a descriptive name like read-only-role. Then, select Create role.

Next, you must assign some permissions to the role. Permissions are represented as permission strings, which have the following pattern: <type>:<action>:<qualified-name>. These permission placeholders are defined as follows:

Select Permissions and enter :read: for the permission string. Then, select the plus (+) icon to add the permission to the role.

Now, we’ll need to create a group to assign this role to. A group is a collection of users who have been grouped together for administrative or security purposes. We’ll create a group and assign the read-only role that you created earlier to it. Then, you can assign users to this group who need read-only access to Decodable, but don’t need to make any updates to it themselves. For example, this can be a group of users who need access to Decodable in order to monitor that their data is making it into the sink.

Select New Group, and give your group a descriptive name like read-only-analysts. Then, select Create group.

After you create the group, you can add users to the group and assign roles to it. If you haven’t already invited the users you want to add to the account, select Account Management > Invite User to send an invitation to join your account. Once the user has accepted, you can add them to the group. When you invite users to join the account, they’re automatically added to the new-users group which has full read and write permissions to the account. Make sure that any users who only need read access to the account are removed from this group.

When you are done adding users to the group, select Roles and assign the role that you created previously to the group.