You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@pinot.apache.org by jl...@apache.org on 2020/10/13 21:56:11 UTC
[incubator-pinot] branch fix-rsvp-meetup created (now 8d40b4a)
This is an automated email from the ASF dual-hosted git repository.
jlli pushed a change to branch fix-rsvp-meetup
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git.
at 8d40b4a Remove tyrus dependencies
This branch includes the following new commits:
new 8d40b4a Remove tyrus dependencies
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org
[incubator-pinot] 01/01: Remove tyrus dependencies
Posted by jl...@apache.org.
This is an automated email from the ASF dual-hosted git repository.
jlli pushed a commit to branch fix-rsvp-meetup
in repository https://gitbox.apache.org/repos/asf/incubator-pinot.git
commit 8d40b4a4706b7123b8da099b149807fdbc32f30e
Author: Jack Li(Analytics Engineering) <jl...@jlli-mn1.linkedin.biz>
AuthorDate: Tue Oct 13 14:55:15 2020 -0700
Remove tyrus dependencies
---
.../org/apache/pinot/tools/RealtimeQuickStart.java | 9 ++-
.../pinot/tools/streams/MeetupRsvpStream.java | 73 +++++++++++++++++++++-
2 files changed, 78 insertions(+), 4 deletions(-)
diff --git a/pinot-tools/src/main/java/org/apache/pinot/tools/RealtimeQuickStart.java b/pinot-tools/src/main/java/org/apache/pinot/tools/RealtimeQuickStart.java
index 0af2b7c..23d06e6 100644
--- a/pinot-tools/src/main/java/org/apache/pinot/tools/RealtimeQuickStart.java
+++ b/pinot-tools/src/main/java/org/apache/pinot/tools/RealtimeQuickStart.java
@@ -38,6 +38,7 @@ import static org.apache.pinot.tools.Quickstart.printStatus;
public class RealtimeQuickStart {
private StreamDataServerStartable _kafkaStarter;
+ private File _dataFile;
public static void main(String[] args)
throws Exception {
@@ -64,6 +65,12 @@ public class RealtimeQuickStart {
com.google.common.base.Preconditions.checkNotNull(resource);
FileUtils.copyURLToFile(resource, tableConfigFile);
+ _dataFile = new File(configDir, "meetupRsvp_data.json");
+ resource = Quickstart.class.getClassLoader()
+ .getResource("examples/stream/meetupRsvp/sample_data/meetupRsvp_data.json");
+ Preconditions.checkNotNull(resource);
+ FileUtils.copyURLToFile(resource, _dataFile);
+
QuickstartTableRequest request = new QuickstartTableRequest("meetupRsvp", schemaFile, tableConfigFile);
final QuickstartRunner runner = new QuickstartRunner(Lists.newArrayList(request), 1, 1, 1, dataDir);
@@ -77,7 +84,7 @@ public class RealtimeQuickStart {
_kafkaStarter.start();
_kafkaStarter.createTopic("meetupRSVPEvents", KafkaStarterUtils.getTopicCreationProps(10));
printStatus(Color.CYAN, "***** Starting meetup data stream and publishing to Kafka *****");
- MeetupRsvpStream meetupRSVPProvider = new MeetupRsvpStream();
+ MeetupRsvpStream meetupRSVPProvider = new MeetupRsvpStream(_dataFile);
meetupRSVPProvider.run();
printStatus(Color.CYAN, "***** Starting Zookeeper, controller, server and broker *****");
runner.startAll();
diff --git a/pinot-tools/src/main/java/org/apache/pinot/tools/streams/MeetupRsvpStream.java b/pinot-tools/src/main/java/org/apache/pinot/tools/streams/MeetupRsvpStream.java
index 86a89db..f42e897 100644
--- a/pinot-tools/src/main/java/org/apache/pinot/tools/streams/MeetupRsvpStream.java
+++ b/pinot-tools/src/main/java/org/apache/pinot/tools/streams/MeetupRsvpStream.java
@@ -20,7 +20,11 @@ package org.apache.pinot.tools.streams;
import com.fasterxml.jackson.databind.JsonNode;
import com.fasterxml.jackson.databind.node.ObjectNode;
+import java.io.BufferedReader;
+import java.io.File;
+import java.io.FileInputStream;
import java.io.IOException;
+import java.io.InputStreamReader;
import java.net.URI;
import java.nio.charset.StandardCharsets;
import java.util.Properties;
@@ -43,13 +47,73 @@ public class MeetupRsvpStream {
private boolean keepPublishing = true;
private ClientManager client;
- public MeetupRsvpStream()
+ private FileInputStream _fileInputStream;
+ private BufferedReader _bufferedReader;
+ private File _dataFile;
+
+ public MeetupRsvpStream(File dataFile)
throws Exception {
Properties properties = new Properties();
properties.put("metadata.broker.list", KafkaStarterUtils.DEFAULT_KAFKA_BROKER);
properties.put("serializer.class", "kafka.serializer.DefaultEncoder");
properties.put("request.required.acks", "1");
producer = StreamDataProvider.getStreamDataProducer(KafkaStarterUtils.KAFKA_PRODUCER_CLASS_NAME, properties);
+ _dataFile = dataFile;
+ }
+
+ public void publishStream()
+ throws IOException {
+
+ _fileInputStream = new FileInputStream(_dataFile);
+ _bufferedReader = new BufferedReader(new InputStreamReader(_fileInputStream, "UTF8"));
+
+ String jsonString;
+ try {
+ while ((jsonString = _bufferedReader.readLine()) != null) {
+ if (!keepPublishing) {
+ return;
+ }
+ System.out.println(jsonString);
+ if (jsonString.isEmpty()) {
+ LOGGER.error("Json string is empty");
+ System.out.println("Json string is empty");
+ continue;
+ }
+ JsonNode messageJSON = JsonUtils.stringToJsonNode(jsonString);
+ ObjectNode extracted = JsonUtils.newObjectNode();
+
+ JsonNode venue = messageJSON.get("venue");
+ if (venue != null) {
+ extracted.set("venue_name", venue.get("venue_name"));
+ }
+
+ JsonNode event = messageJSON.get("event");
+ if (event != null) {
+ extracted.set("event_name", event.get("event_name"));
+ extracted.set("event_id", event.get("event_id"));
+ extracted.set("event_time", event.get("time"));
+ }
+
+ JsonNode group = messageJSON.get("group");
+ if (group != null) {
+ extracted.set("group_city", group.get("group_city"));
+ extracted.set("group_country", group.get("group_country"));
+ extracted.set("group_id", group.get("group_id"));
+ extracted.set("group_name", group.get("group_name"));
+ extracted.set("group_lat", group.get("group_lat"));
+ extracted.set("group_lon", group.get("group_lon"));
+ }
+
+ extracted.set("mtime", messageJSON.get("mtime"));
+ extracted.put("rsvp_count", 1);
+
+ if (keepPublishing) {
+ producer.produce("meetupRSVPEvents", extracted.toString().getBytes(StandardCharsets.UTF_8));
+ }
+ }
+ } catch (Exception e) {
+ LOGGER.error("encountered an error running the meetupRSVPEvents stream", e);
+ }
}
public void stopPublishing() {
@@ -58,7 +122,9 @@ public class MeetupRsvpStream {
client.shutdown();
}
- public void run() {
+ public void run()
+ throws IOException {
+// publishStream();
try {
ClientEndpointConfig cec = ClientEndpointConfig.Builder.create().build();
client = ClientManager.createClient();
@@ -72,6 +138,7 @@ public class MeetupRsvpStream {
@Override
public void onMessage(String message) {
try {
+ LOGGER.info("AAA: " + message);
JsonNode messageJSON = JsonUtils.stringToJsonNode(message);
ObjectNode extracted = JsonUtils.newObjectNode();
@@ -113,7 +180,7 @@ public class MeetupRsvpStream {
LOGGER.error("found an event where data did not have all the fields, don't care about for quickstart", e);
}
}
- }, cec, new URI("ws://stream.meetup.com/2/rsvps"));
+ }, cec, new URI("http://stream.meetup.com/2/rsvps"));
} catch (Exception e) {
LOGGER.error("encountered an error running the meetupRSVPEvents stream", e);
}
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@pinot.apache.org
For additional commands, e-mail: commits-help@pinot.apache.org