Connect and Upload Data
Uploading directly from
Python
Data can be uploaded via
the Python API. If your data exists in HDFS, Kafka, or another data source, it
may be preferable to use connectors.
Streaming
Data
Create a
Stream
Create a named stream to
start sending events to it. By default a stream will retain its events forever;
however, its retention behavior may be changed with the
retention_policy parameter.
1
stream =
streamsql.source.create_stream(name="MyStream")
Copied!
Upload
Event
Events may be uploaded to
the stream. Their timestamp is set to the time the event reaches the StreamSQL
server. By default, the events payload is converted into JSON, but you may
specify the formatting.
1
stream.upload({"key": "value"}, format="json")
Copied!
Dynamic
Tables
Mutable tables are
schema-less keyed tables. Items may be inserted and retrieved by key, and the
table may be projected into another mutable table using a transformation.
1
table =
streamsql.source.create_table(
2
name="MyTable",
3
primary_key="id",
4
)
5
table.insert({"id": "user1", "first_name": "simba", "last_name": "khadder"})
6
table.get("user1")
Copied!
Dynamic Tables may not
be used in joins and may not be versioned.
Immutable
Tables
Like Dynamic
Tables, Immutable Tables are
schema-less keyed tables. They may be initialized with all of their data
included. They may also be initialized via a dynamic table then turned into
immutable with
table.make_immutable(version=1). If the data exists in a
file or database table, you should use the connector
APIs.
1
table =
streamsql.source.create_table(
2
name="MyTable",
3
primary_key="id",
4
data=[{"id": "user1", "first_name": "simba", "last_name": "khadder"}].
5
immutable=True,
6
version=1,
7
)
8
table.get("user1")
Copied!
Connecting External
Data Sources
Files
Immutable tables may be
created from files. In this case, the file is pulled and copied into StreamSQL.
You can create the same table with a new version if the file
changes.
1
streamsql.connector.upload_file(
2
"hdfs://data/users.csv",
3
name="users",
4
primary_key="id",
5
)
Copied!
Streams
Kafka streams may be
uploaded directly into a StreamSQL stream. In this case, we will start reading
from Kafka at its earliest messages and retain all messages indefinitely in
StreamSQL to power our backfilling.
1
streamsql.connector.upload_kafka(
2
brokers=["localhost:9092"],
3
name="transactions",
4
format="json",
5
)
Copied!
Last
modified 1yr
ago