Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Added spark streaming custom receiver for pulsar #296

Merged
merged 2 commits into from
Apr 4, 2017
Merged
Show file tree
Hide file tree
Changes from 1 commit
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
1 change: 1 addition & 0 deletions docs/Documentation.md
Original file line number Diff line number Diff line change
Expand Up @@ -12,6 +12,7 @@
- [Geo-Replication](GeoReplication.md)
- [WebSocket API](WebSocket.md)
- [Apache Storm adaptor](PulsarStorm.md)
- [Spark Streaming Pulsar Receiver](PulsarSpark.md)
* Internal Docs
- [Binary protocol specification](BinaryProtocol.md)
* Other Languages
Expand Down
45 changes: 45 additions & 0 deletions docs/PulsarSpark.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,45 @@
# Spark Streaming Pulsar Receiver

<!-- TOC depthFrom:2 depthTo:3 withLinks:1 updateOnSave:1 orderedList:0 -->

- [Introduction](#introduction)
- [Using Spark Streaming Pulsar Receiver](#using-spark-streaming-pulsar-receiver)
- [Example](#example)

<!-- /TOC -->

## Introduction
Spark Streaming Pulsar Receiver is a custom receiver which enables Apache Spark Streaming to receive data from Pulsar.

An application can receive RDD format data via Spark Streaming Pulsar Receiver and can process it variously.

## Using Spark Streaming Pulsar Receiver
Include dependency for Spark Streaming Pulsar Receiver:

```xml
<dependency>
<groupId>com.yahoo.pulsar</groupId>
<artifactId>pulsar-spark</artifactId>
<version>${pulsar.version}</version>
</dependency>
```

Pass an instance of SparkStreamingPulsarReceiver to receiverStream method in JavaStreamingContext:
```java
SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("pulsar-spark");
JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(5));

ClientConfiguration clientConf = new ClientConfiguration();
ConsumerConfiguration consConf = new ConsumerConfiguration();
String url = "pulsar://localhost:6650/";
String topic = "persistent://sample/standalone/ns1/topic1";
String subs = "sub1";

JavaReceiverInputDStream<byte[]> msgs = jssc
.receiverStream(new SparkStreamingPulsarReceiver(clientConf, consConf, url, topic, subs));
```


## Example
You can find a complete example [here](../pulsar-spark/src/test/java/com/yahoo/pulsar/spark/example/SparkStreamingPulsarReceiverExample.java).
In this example, the number of messages which contain "Pulsar" string in received messages is counted.
1 change: 1 addition & 0 deletions docs/locale/ja/Documentation.md
Original file line number Diff line number Diff line change
Expand Up @@ -11,5 +11,6 @@
- [ジオレプリケーション](GeoReplication.md)
- [WebSocket API](WebSocket.md)
- [Apache Stormのためのアダプタ](PulsarStorm.md)
- [Spark Streaming Pulsar Receiver](PulsarSpark.md)
* 内部仕様
- [バイナリプロトコルの仕様](BinaryProtocol.md)
45 changes: 45 additions & 0 deletions docs/locale/ja/PulsarSpark.md
Original file line number Diff line number Diff line change
@@ -0,0 +1,45 @@
# Spark Streaming Pulsar Receiver

<!-- TOC depthFrom:2 depthTo:3 withLinks:1 updateOnSave:1 orderedList:0 -->

- [概要](#概要)
- [Spark Streaming Pulsar Receiverの利用](#spark-streaming-pulsar-receiverの利用)
- [実装例](#実装例)

<!-- /TOC -->

## 概要
Spark Streaming Pulsar ReceiverはApache Spark StreamingがPulsarからデータを受け取るためのCustom Receiverです。

アプリケーションはSpark Streaming Pulsar Receiverを通してRDD形式のデータを受け取り、様々な処理を行うことができます。

## Spark Streaming Pulsar Receiverの利用
Spark Streaming Pulsar Receiverの依存をincludeします:

```xml
<dependency>
<groupId>com.yahoo.pulsar</groupId>
<artifactId>pulsar-spark</artifactId>
<version>${pulsar.version}</version>
</dependency>
```

JavaStreamingContextのreceiverStreamメソッドにSparkStreamingPulsarReceiverのインスタンスを渡します:
```java
SparkConf conf = new SparkConf().setMaster("local[*]").setAppName("pulsar-spark");
JavaStreamingContext jssc = new JavaStreamingContext(conf, Durations.seconds(5));

ClientConfiguration clientConf = new ClientConfiguration();
ConsumerConfiguration consConf = new ConsumerConfiguration();
String url = "pulsar://localhost:6650/";
String topic = "persistent://sample/standalone/ns1/topic1";
String subs = "sub1";

JavaReceiverInputDStream<byte[]> msgs = jssc
.receiverStream(new SparkStreamingPulsarReceiver(clientConf, consConf, url, topic, subs));
```


## 実装例
完全な実装の例は[SparkStreamingPulsarReceiver.java](../../../pulsar-spark/src/test/java/com/yahoo/pulsar/spark/example/SparkStreamingPulsarReceiverExample.java)を参照してください。
この例では、受け取ったメッセージのうち"Pulsar"という文字列が含まれるものがいくつあるかを数えます。
29 changes: 29 additions & 0 deletions pom.xml
Original file line number Diff line number Diff line change
Expand Up @@ -78,6 +78,7 @@ flexible messaging model and an intuitive client API.</description>
<module>pulsar-websocket</module>
<module>pulsar-discovery-service</module>
<module>pulsar-storm</module>
<module>pulsar-spark</module>
<module>pulsar-zookeeper-utils</module>
<module>pulsar-checksum</module>
<module>pulsar-testclient</module>
Expand Down Expand Up @@ -404,6 +405,34 @@ flexible messaging model and an intuitive client API.</description>
<version>${athenz.version}</version>
</dependency>

<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.10</artifactId>
<version>2.1.0</version>
<exclusions>
<exclusion>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
</exclusion>
<exclusion>
<groupId>io.netty</groupId>
<artifactId>netty-codec-http</artifactId>
</exclusion>
<exclusion>
<groupId>io.netty</groupId>
<artifactId>netty-transport-native-epoll</artifactId>
</exclusion>
<exclusion>
<groupId>io.netty</groupId>
<artifactId>netty</artifactId>
</exclusion>
<exclusion>
<groupId>io.netty</groupId>
<artifactId>netty-all</artifactId>
</exclusion>
</exclusions>
</dependency>

</dependencies>
</dependencyManagement>

Expand Down
124 changes: 124 additions & 0 deletions pulsar-spark/pom.xml
Original file line number Diff line number Diff line change
@@ -0,0 +1,124 @@
<!--

Copyright 2016 Yahoo Inc.

Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at

http://www.apache.org/licenses/LICENSE-2.0

Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.

-->
<project
xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd"
xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance">
<modelVersion>4.0.0</modelVersion>

<parent>
<groupId>com.yahoo.pulsar</groupId>
<artifactId>pulsar</artifactId>
<version>1.17-SNAPSHOT</version>
<relativePath>..</relativePath>
</parent>

<artifactId>pulsar-spark</artifactId>
<name>Spark Streaming Pulsar Receivers</name>

<dependencies>
<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>pulsar-broker</artifactId>
<version>${project.version}</version>
<scope>test</scope>
</dependency>

<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>pulsar-broker</artifactId>
<version>${project.version}</version>
<scope>test</scope>
<type>test-jar</type>
</dependency>

<dependency>
<groupId>org.mockito</groupId>
<artifactId>mockito-core</artifactId>
<scope>test</scope>
</dependency>

<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>managed-ledger</artifactId>
<version>${project.version}</version>
<scope>test</scope>
<type>test-jar</type>
</dependency>

<dependency>
<groupId>${project.groupId}</groupId>
<artifactId>pulsar-client</artifactId>
<version>${project.version}</version>
</dependency>

<dependency>
<groupId>org.slf4j</groupId>
<artifactId>slf4j-api</artifactId>
</dependency>

<dependency>
<groupId>org.apache.spark</groupId>
<artifactId>spark-streaming_2.10</artifactId>
</dependency>

<dependency>
<groupId>com.google.guava</groupId>
<artifactId>guava</artifactId>
</dependency>

</dependencies>
<build>
<plugins>
<plugin>
<groupId>org.apache.maven.plugins</groupId>
<artifactId>maven-shade-plugin</artifactId>
<executions>
<execution>
<phase>package</phase>
<goals>
<goal>shade</goal>
</goals>
<configuration>
<createDependencyReducedPom>true</createDependencyReducedPom>
<promoteTransitiveDependencies>ture</promoteTransitiveDependencies>
<artifactSet>
<includes>
<include>com.google.guava:guava</include>
<include>io.netty:netty-codec-http</include>
<include>io.netty:netty-transport-native-epoll</include>
<include>io.netty:netty</include>
<include>io.netty:netty-all</include>
</includes>
</artifactSet>
<relocations>
<relocation>
<pattern>com.google</pattern>
<shadedPattern>com.yahoo.pulsar.shade.com.google</shadedPattern>
</relocation>
<relocation>
<pattern>io.netty</pattern>
<shadedPattern>com.yahoo.pulsar.shade.io.netty</shadedPattern>
</relocation>
</relocations>
</configuration>
</execution>
</executions>
</plugin>
</plugins>
</build>
</project>
Original file line number Diff line number Diff line change
@@ -0,0 +1,91 @@
/**
* Copyright 2016 Yahoo Inc.
*
* Licensed under the Apache License, Version 2.0 (the "License");
* you may not use this file except in compliance with the License.
* You may obtain a copy of the License at
*
* http://www.apache.org/licenses/LICENSE-2.0
*
* Unless required by applicable law or agreed to in writing, software
* distributed under the License is distributed on an "AS IS" BASIS,
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
* See the License for the specific language governing permissions and
* limitations under the License.
*/

package com.yahoo.pulsar.spark;

import org.apache.spark.storage.StorageLevel;
import org.apache.spark.streaming.receiver.Receiver;

import com.yahoo.pulsar.client.api.ClientConfiguration;
import com.yahoo.pulsar.client.api.Consumer;
import com.yahoo.pulsar.client.api.ConsumerConfiguration;
import com.yahoo.pulsar.client.api.Message;
import com.yahoo.pulsar.client.api.MessageListener;
import com.yahoo.pulsar.client.api.PulsarClient;
import com.yahoo.pulsar.client.api.PulsarClientException;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.concurrent.TimeUnit;

public class SparkStreamingPulsarReceiver extends Receiver<byte[]> {

private ClientConfiguration clientConfiguration;
private ConsumerConfiguration consumerConfiguration;
private PulsarClient pulsarClient;
private String url;
private String topic;
private String subscription;

public SparkStreamingPulsarReceiver(ClientConfiguration clientConfiguration,
ConsumerConfiguration consumerConfiguration, String url, String topic, String subscription) {
this(StorageLevel.MEMORY_AND_DISK_2(), clientConfiguration, consumerConfiguration, url, topic, subscription);
}

public SparkStreamingPulsarReceiver(StorageLevel storageLevel, ClientConfiguration clientConfiguration,
ConsumerConfiguration consumerConfiguration, String url, String topic, String subscription) {
super(storageLevel);
this.clientConfiguration = clientConfiguration;
this.url = url;
this.topic = topic;
this.subscription = subscription;
if (consumerConfiguration.getAckTimeoutMillis() == 0) {
consumerConfiguration.setAckTimeout(60, TimeUnit.SECONDS);
}
consumerConfiguration.setMessageListener((consumer, msg) -> {
try {
store(msg.getData());
consumer.acknowledgeAsync(msg);
} catch (Exception e) {
log.error("Failed to store a message : {}", e.getMessage());
}
});
this.consumerConfiguration = consumerConfiguration;
}

public void onStart() {
try {
pulsarClient = PulsarClient.create(url, clientConfiguration);
pulsarClient.subscribe(topic, subscription, consumerConfiguration);
} catch (PulsarClientException e) {
log.error("Failed to start subscription : {}", e.getMessage());
restart("Restart a consumer");
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Wouldn't be possible to just propagate the exception at this point?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

PulsarClientException is so a checked exception that it has to be caught here.
If we try to propagate like following, onStart() cannot be overrode.

public void onStart() throws PulsarClientExceptio{
catch (PulsarClientException e) {
    throw new PulsarClientException("error!");
}

}
}

public void onStop() {
try {
if (pulsarClient != null) {
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

is this transition valid? onStart -> onStop -> onStart?

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

onStart and onStop are called only once.
onStart -> onStop

Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

ok. the reason i asked is that it's now a bit asymmetric after you made the change i suggested to onStart(): onStop() invalidates the invariants that onStart() relies upon.

Copy link
Contributor Author

@yush1ga yush1ga Mar 22, 2017

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Sorry, I confirmed that when restart() is called, onStop() -> onStart() is called.
As we create a client in the constructor, it cannot start subscribing when restarting.
I think we should create a client in onStart().

pulsarClient.close();
}
} catch (PulsarClientException e) {
log.error("Failed to close client : {}", e.getMessage());
}
}

private static final Logger log = LoggerFactory.getLogger(SparkStreamingPulsarReceiver.class);
}
Copy link
Contributor

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

We should also update documentation .

Copy link
Contributor Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

OK. I'll make it.

Loading