Goal:
KSQL is a streaming SQL engine that enables stream processing against Apache Kafka and MapR Streams(aka MapR-ES). KSQL interacts directly with the Kafka Streams API, removing the requirement of building a Java app.This article provides an example to quickly help you hands on ksql on MapR Streams by referring to this quick start.
Env:
MapR 6.0.1Solution:
Note: ksql is still in developer preview in MapR 6.0.1.1. Download the ksql tarball.
wget http://package.mapr.com/labs/ksql/package-ksql-4.1.0-SNAPSHOT-package.tgz tar -xzvf package-ksql-4.1.0-SNAPSHOT-package.tgz cd ksql
2. Create a MapR intenal Stream with public access
This is because security is not supported for this Developer Preview on MapR 6.0.1.maprcli stream create -path /var/mapr/kafka-internal-stream -produceperm p -consumeperm p -topicperm p
- It is used for internal topic creation and cli-side partition assignment.
- Internal stream should be created before starting KSQL server.
[root@v2 data]# maprcli stream topic list -path /var/mapr/kafka-internal-stream partitions maxlag logicalsize topic consumers physicalsize 1 0 0 __mapr__ksql_transient_4130383735097621206_1536269653676_assignment 0 0 1 0 0 ksql__commands 0 0 1 0 0 __mapr__ksql_transient_7811816849950865146_1536269574284_assignment 0 0
3. Modify the configuration for ksql
vim ./config/ksqlserver.propertiesSample is:
bootstrap.servers=localhost:9092 ksql.command.topic.suffix=commands listeners=http://localhost:8080 application.id=myksql
4. Create a sample MapR stream to work with for below use case.
maprcli stream create -path /sample-stream -produceperm p -consumeperm p -topicperm p
5. Generate messages using ksql-datagen
ksql provides a very handy tool named "ksql-datagen" to help avoid creating a test producer program.Refer to this link.
Here we are generating sample messages into 2 topics named "pageviews" and "users".
The schema is shown as below:
- "pageviews":
./bin/ksql-datagen quickstart=pageviews format=delimited topic=/sample-stream:pageviews maxInterval=10000sample data is like:
1 --> ([ 1536339899430 | 'User_8' | 'Page_12' ]) 11 --> ([ 1536339904639 | 'User_9' | 'Page_11' ]) 21 --> ([ 1536339906898 | 'User_1' | 'Page_25' ]) ...
- "users":
./bin/ksql-datagen quickstart=users format=json topic=/sample-stream:users maxInterval=10000sample data is like:
User_6 --> ([ 1488285478109 | 'User_6' | 'Region_1' | 'FEMALE' ]) User_7 --> ([ 1509065238387 | 'User_7' | 'Region_5' | 'OTHER' ]) User_2 --> ([ 1491876978482 | 'User_2' | 'Region_1' | 'OTHER' ]) ...Then use below 2 commands to check the topic information inside the MapR Stream:
maprcli stream topic info -path /sample-stream -topic pageviews -json maprcli stream topic info -path /sample-stream -topic users -json
6. Start ksql in standalone mode
./bin/ksql-cli local --properties-file ./config/ksqlserver.propertiesAfter that, the ksql java process is running:
[root@v2 config]# ps -ef|grep -i ksql root 6768 13847 66 12:08 pts/1 00:00:07 java -cp /root/ksql/share/java/ksql/*: -Xmx3g -server -XX:+UseParNewGC -XX:+UseConcMarkSweepGC -XX:+CMSClassUnloadingEnabled -XX:+CMSScavengeBeforeRemark -XX:+DisableExplicitGC -Djava.awt.headless=true -Dcom.sun.management.jmxremote -Dcom.sun.management.jmxremote.authenticate=false -Dcom.sun.management.jmxremote.ssl=false -Dksql.log.dir=/tmp/ksql-logs -Dlog4j.configuration=file:/root/ksql/config/log4j-file.properties io.confluent.ksql.Ksql local --properties-file ./config/ksqlserver.propertiesShow the current properties:
ksql> SHOW PROPERTIES; Property | Value ------------------------------------------------------------------------------------------------------------------ streams.default.stream | ksql.transient.prefix | transient_ commit.interval.ms | 2000 ksql.sink.replicas | 1 listeners | http://localhost:9098 default.deserialization.exception.handler | io.confluent.ksql.errors.LogMetricAndContinueExceptionHandler bootstrap.servers | localhost:9099 ksql.schema.registry.url | http://localhost:8081 ksql.sink.partitions | 4 ksql.statestore.suffix | _ksql_statestore ksql.service.id | ksql_ cache.max.bytes.buffering | 10000000 ksql.default.stream | ksql.sink.window.change.log.additional.retention | 1000000 auto.offset.reset | latest num.stream.threads | 4 ksql.persistent.prefix | query_ application.id | ksql_ ------------------------------------------------------------------------------------------------------------------
7. Create stream for topic "pageviews" and create table for topic "users".
Here the "stream" and "table" are concept of kafka.- A stream can be considered a changelog of a table, where each data record in the stream captures a state change of the table. Like an "INSERT".
- A table can be considered a snapshot, at a point in time, of the latest value for each key in a stream (a stream’s data records are key-value pairs). Like an "UPDATE".
Now here are the exact commands in ksql:
ksql> CREATE STREAM pageviews_original (viewtime bigint, userid varchar, pageid varchar) WITH (kafka_topic='/sample-stream:pageviews', value_format='DELIMITED'); Message ---------------- Stream created ---------------- ksql> DESCRIBE pageviews_original; Field | Type -------------------------------------- ROWTIME | BIGINT (system) ROWKEY | VARCHAR(STRING) (system) VIEWTIME | BIGINT USERID | VARCHAR(STRING) PAGEID | VARCHAR(STRING) -------------------------------------- For runtime statistics and query details run: DESCRIBE EXTENDED <Stream,Table>; ksql> CREATE TABLE users_original (registertime bigint, gender varchar, regionid varchar, userid varchar) WITH (kafka_topic='/sample-stream:users', value_format='JSON', key = 'userid'); Message --------------- Table created --------------- ksql> DESCRIBE users_original; Field | Type ------------------------------------------ ROWTIME | BIGINT (system) ROWKEY | VARCHAR(STRING) (system) REGISTERTIME | BIGINT GENDER | VARCHAR(STRING) REGIONID | VARCHAR(STRING) USERID | VARCHAR(STRING) ------------------------------------------ For runtime statistics and query details run: DESCRIBE EXTENDED <Stream,Table>;
8. Running a query reading above stream or table.
By default KSQL reads the topics for streams and tables from the latest offset.ksql> select * from pageviews_original ; 1536342236738 | 4871 | 1536342235234 | User_8 | Page_12 1536342243408 | 4881 | 1536342241904 | User_4 | Page_18 1536342252370 | 4891 | 1536342250867 | User_4 | Page_13 ...The default location for local state store (RocksDB) is /tmp/kafka-streams, and for each KStreams application subdirectory /tmp/kafka-streams/<application.id> is created.
In the meantime, we can check the application id for the current running queries:
[root@v2 ~]# ls -altr /tmp/kafka-streams/ drwxr-xr-x 3 root root 17 Sep 7 10:54 ksql_transient_1376501806109265799_1536342862444Then using the application.id we can check the "jstack <ksql pid>" output to find out the related threads for this query:
[root@v2 ~]# jstack 6768 |grep ksql_transient_1376501806109265799_1536342862444 "ksql_transient_1376501806109265799_1536342862444-76a58c09-c3a3-4c13-adb7-ab00b7f7ccf7-CleanupThread" #197 daemon prio=5 os_prio=0 tid=0x00007f9298076800 nid=0x1ed4 waiting on condition [0x00007f927d0e3000] "ksql_transient_1376501806109265799_1536342862444-76a58c09-c3a3-4c13-adb7-ab00b7f7ccf7-StreamThread-8" #196 prio=5 os_prio=0 tid=0x00007f9298075000 nid=0x1ed3 runnable [0x00007f927d1e4000] "ksql_transient_1376501806109265799_1536342862444-76a58c09-c3a3-4c13-adb7-ab00b7f7ccf7-StreamThread-7" #195 prio=5 os_prio=0 tid=0x00007f9298074000 nid=0x1ed2 runnable [0x00007f927d3e6000] "ksql_transient_1376501806109265799_1536342862444-76a58c09-c3a3-4c13-adb7-ab00b7f7ccf7-StreamThread-6" #194 prio=5 os_prio=0 tid=0x00007f9298073000 nid=0x1ed1 runnable [0x00007f927d4e7000] "ksql_transient_1376501806109265799_1536342862444-76a58c09-c3a3-4c13-adb7-ab00b7f7ccf7-StreamThread-5" #193 prio=5 os_prio=0 tid=0x00007f92982e9800 nid=0x1ed0 runnable [0x00007f927dced000]
9. Create a persistent query to store the data into a MapR Stream topic
Unlike the non-persistent query above, results from this query are written to a MapR Stream topic "pageviews_female".CREATE STREAM pageviews_female WITH (kafka_topic='/sample-stream:pageviews_female', value_format='json') AS SELECT users_original.userid AS userid, pageid, regionid, gender FROM pageviews_original LEFT JOIN users_original ON pageviews_original.userid = users_original.userid WHERE gender = 'FEMALE';After the stream is created and returned, actually the thread is running in the background:
[root@v2 ~]# jstack 6768 |grep ksql_query_CSAS_PAGEVIEWS_FEMALE "ksql_query_CSAS_PAGEVIEWS_FEMALE-af068d1f-9ff4-4acb-abe7-e4ff37a5c91c-CleanupThread" #261 daemon prio=5 os_prio=0 tid=0x00007f92e0039000 nid=0x27aa waiting on condition [0x00007f927cee1000] "ksql_query_CSAS_PAGEVIEWS_FEMALE-af068d1f-9ff4-4acb-abe7-e4ff37a5c91c-StreamThread-16" #260 prio=5 os_prio=0 tid=0x00007f92e0037800 nid=0x27a9 runnable [0x00007f927cfe2000] "ksql_query_CSAS_PAGEVIEWS_FEMALE-af068d1f-9ff4-4acb-abe7-e4ff37a5c91c-StreamThread-15" #259 prio=5 os_prio=0 tid=0x00007f92e0037000 nid=0x27a8 runnable [0x00007f927d0e3000] "ksql_query_CSAS_PAGEVIEWS_FEMALE-af068d1f-9ff4-4acb-abe7-e4ff37a5c91c-StreamThread-14" #258 prio=5 os_prio=0 tid=0x00007f92e0036000 nid=0x27a7 runnable [0x00007f927d1e4000] "ksql_query_CSAS_PAGEVIEWS_FEMALE-af068d1f-9ff4-4acb-abe7-e4ff37a5c91c-StreamThread-13" #257 prio=5 os_prio=0 tid=0x00007f92e002d000 nid=0x27a6 runnable [0x00007f927dced000]And the MapR Stream topic is created:
[root@v2 ~]# maprcli stream topic info -path /sample-stream -topic pageviews_female -json
{
"timestamp":1536344466684,
"timeofday":"2018-09-07 11:21:06.684 GMT-0700 AM",
"status":"OK",
"total":4,
"data":[
{
"partitionid":0,
"physicalsize":0,
"logicalsize":0,
"maxoffset":10,
"minoffsetacrossconsumers":0,
"mintimestamp":"2018-09-07T11:16:20.523-0700 AM",
"maxtimestamp":"2018-09-07T11:21:00.328-0700 AM",
"mintimestampacrossconsumers":"1969-12-31T04:00:00.000-0800 PM",
"fid":"2078.69.262670",
"master":"v1.poc.com:5660",
"servers":"v2.poc.com:5660, v1.poc.com:5660, v4.poc.com:5660",
"timestamptype":"CreateTime"
},
{
"partitionid":1,
"physicalsize":212992,
"logicalsize":90112,
"maxoffset":4,
"minoffsetacrossconsumers":0,
"mintimestamp":"2018-09-07T11:19:19.452-0700 AM",
"maxtimestamp":"2018-09-07T11:20:57.326-0700 AM",
"mintimestampacrossconsumers":"1969-12-31T04:00:00.000-0800 PM",
"fid":"2092.32.262366",
"master":"v2.poc.com:5660",
"servers":"v2.poc.com:5660, v4.poc.com:5660, v3.poc.com:5660",
"timestamptype":"CreateTime"
},
{
"partitionid":2,
"physicalsize":0,
"logicalsize":0,
"maxoffset":3,
"minoffsetacrossconsumers":0,
"mintimestamp":"2018-09-07T11:18:47.229-0700 AM",
"maxtimestamp":"2018-09-07T11:20:52.785-0700 AM",
"mintimestampacrossconsumers":"1969-12-31T04:00:00.000-0800 PM",
"fid":"2182.96.262582",
"master":"v4.poc.com:5660",
"servers":"v1.poc.com:5660, v4.poc.com:5660, v3.poc.com:5660",
"timestamptype":"CreateTime"
},
{
"partitionid":3,
"physicalsize":0,
"logicalsize":0,
"maxoffset":6,
"minoffsetacrossconsumers":0,
"mintimestamp":"2018-09-07T11:15:48.273-0700 AM",
"maxtimestamp":"2018-09-07T11:20:13.588-0700 AM",
"mintimestampacrossconsumers":"1969-12-31T04:00:00.000-0800 PM",
"fid":"2080.41.262504",
"master":"v1.poc.com:5660",
"servers":"v2.poc.com:5660, v1.poc.com:5660, v3.poc.com:5660",
"timestamptype":"CreateTime"
}
]
}
Since here the result topic "pageviews_female" is created in json format, so basically you can also use Drill to query it as well. Refer to this link.
CFDE0E3696
ReplyDeleteGörüntülü Seks
Görüntülü Show
Görüntülü Show Uygulamaları
13BD7D87
ReplyDeleteDüzce Esçort
Niğde Esçort
Çanakkale Esçort
Elazığ Esçort
Uşak Esçort
Mardin Esçort
Giresun Esçort
Sivas Esçort
Yalova Esçort
2341A289
ReplyDeleteमीटफ्लाईरैंडम चैट
chat casuale
Zufälliger Chat
צ'אט אקראי
Obrolan Acak
ميت فلايدردشة عشوائية
Random Chat
Willekeurige chat
Náhodný chat