-
Notifications
You must be signed in to change notification settings - Fork 32
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
Merge pull request #119 from Alex-Cook4/master
Adding sample and fixing makefiles
- Loading branch information
Showing
17 changed files
with
62,718 additions
and
15 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,11 @@ | ||
| <?xml version="1.0" encoding="UTF-8"?> | ||
| <classpath> | ||
| <classpathentry exported="true" kind="con" path="com.ibm.streams.java/com.ibm.streams.operator"/> | ||
| <classpathentry exported="true" kind="con" path="org.eclipse.jdt.launching.JRE_CONTAINER"/> | ||
| <classpathentry kind="src" path=".apt_generated"> | ||
| <attributes> | ||
| <attribute name="optional" value="true"/> | ||
| </attributes> | ||
| </classpathentry> | ||
| <classpathentry kind="output" path="output"/> | ||
| </classpath> |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,29 @@ | ||
| <?xml version="1.0" encoding="UTF-8"?> | ||
| <projectDescription> | ||
| <name>KafkaParallelConsumers</name> | ||
| <comment></comment> | ||
| <projects> | ||
| </projects> | ||
| <buildSpec> | ||
| <buildCommand> | ||
| <name>org.eclipse.jdt.core.javabuilder</name> | ||
| <arguments> | ||
| </arguments> | ||
| </buildCommand> | ||
| <buildCommand> | ||
| <name>com.ibm.streams.studio.splproject.builder.SPLProjectBuilder</name> | ||
| <arguments> | ||
| </arguments> | ||
| </buildCommand> | ||
| <buildCommand> | ||
| <name>org.eclipse.xtext.ui.shared.xtextBuilder</name> | ||
| <arguments> | ||
| </arguments> | ||
| </buildCommand> | ||
| </buildSpec> | ||
| <natures> | ||
| <nature>com.ibm.streams.studio.splproject.SPLProjectNature</nature> | ||
| <nature>org.eclipse.xtext.ui.shared.xtextNature</nature> | ||
| <nature>org.eclipse.jdt.core.javanature</nature> | ||
| </natures> | ||
| </projectDescription> |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,31 @@ | ||
| # Copyright (C) 2015, International Business Machines Corporation | ||
| # All Rights Reserved | ||
|
|
||
| .PHONY: all clean | ||
|
|
||
| TOOLKIT_NAME=com.ibm.streamsx.messaging | ||
| STREAMS_MESSAGING_TOOLKIT ?= $(shell ([ -e ../../$(TOOLKIT_NAME)/toolkit.xml ] && echo ../../$(TOOLKIT_NAME)) ||\ | ||
| echo $(STREAMS_STUDIO_SPL_PATH) ||\ | ||
| ([ -e "../$(TOOLKIT_NAME)" ] && echo ../$(TOOLKIT_NAME)) ||\ | ||
| echo $(STREAMS_INSTALL)/toolkits/$(TOOLKIT_NAME)) | ||
|
|
||
|
|
||
| SPLC_FLAGS ?= -a --data-directory data | ||
| SPLC = $(STREAMS_INSTALL)/bin/sc | ||
|
|
||
| SPL_CMD_ARGS ?= -t $(STREAMS_MESSAGING_TOOLKIT) | ||
| SPL_MAIN_COMPOSITE = application::ParallelConsumers | ||
|
|
||
| all: distributed | ||
|
|
||
| data: | ||
| mkdir data | ||
|
|
||
| standalone: data | ||
| $(SPLC) $(SPLC_FLAGS) -T -M $(SPL_MAIN_COMPOSITE) $(SPL_CMD_ARGS) | ||
|
|
||
| distributed: data | ||
| $(SPLC) $(SPLC_FLAGS) -M $(SPL_MAIN_COMPOSITE) $(SPL_CMD_ARGS) | ||
|
|
||
| clean: | ||
| $(SPLC) $(SPLC_FLAGS) -C -M $(SPL_MAIN_COMPOSITE) |
Empty file.
69 changes: 69 additions & 0 deletions
69
samples/KafkaParallelConsumers/application/KafkaParallelConsumers.spl
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,69 @@ | ||
| namespace application ; | ||
|
|
||
| use com.ibm.streamsx.messaging.kafka::* ; | ||
| /** | ||
| * Read from a three-partition Kafka topic using the KafkaConsumer operator in a parallel region | ||
| * 3-wide. Kafka only guarantees ordering of tuples within a single partition. This application | ||
| * provides the same guarantee, but since we are reading from three separate partitions, | ||
| * we can lose overall topic order. Depending on the key/message, order can be recovered after consuming | ||
| * from a Kafka Server similar to how it is done here: https://developer.ibm.com/streamsdev/docs/parallelized-file-processing-parse-operator/ | ||
| * | ||
| * | ||
| * Make sure you have created your topic before launching: | ||
| * bin/kafka-topics.sh --create --zookeeper <zk.Host.1>:2181 --partitions 3 --topic myParallelTopic | ||
| * | ||
| * Edit the consumer.properties and producer.properties files found in the etc directory to include | ||
| * your Kafka properties. | ||
| * | ||
| * Build using Studio or the provided Makefile. | ||
| * | ||
| * Check results by looking at messagesReceived.out in the data directory. | ||
| * | ||
| */ | ||
| composite ParallelConsumers | ||
| { | ||
| graph | ||
| //generate data to be written to a kafka server | ||
| stream<rstring topic, rstring key, rstring message> OutputStream = Beacon() | ||
| { | ||
| param | ||
| period : 0.25 ; | ||
| initDelay : 4.0 ; | ||
| output | ||
| OutputStream : topic = "myParallelTopic", message =(rstring) | ||
| IterationCount(), key =(rstring)(int32)(random() * 10.0) ; | ||
| } | ||
|
|
||
| //Write to Kafka Server | ||
| () as KafkaSinkOp = KafkaProducer(OutputStream) | ||
| { | ||
| param | ||
| propertiesFile : "etc/producer.properties" ; | ||
| } | ||
|
|
||
| //Read in from a kafka server with a 3-partition topic and start consistent region | ||
| @parallel(width = 3) | ||
| stream<rstring message, rstring key> KafkaConsumerOut = KafkaConsumer() | ||
| { | ||
| param | ||
| propertiesFile : "etc/consumer.properties" ; | ||
| topic : "myParallelTopic" ; | ||
| partition : getChannel() ; | ||
| } | ||
|
|
||
|
|
||
| //Print out data to a file | ||
| () as MessagePrinter = FileSink(KafkaConsumerOut) | ||
| { | ||
| param | ||
| file : "messagesReceived.out" ; | ||
| flush : 1u ; | ||
| format : csv ; | ||
| } | ||
|
|
||
| () as JCP = JobControlPlane() | ||
| { | ||
| } | ||
|
|
||
| } | ||
|
|
Oops, something went wrong.