Skip to content

Commit 6d6776a

Browse files
harishreedharanharishreedharan
committed
SPARK-1729. Make Flume pull data from source, rather than the current push model
Currently Spark uses Flume's internal Avro Protocol to ingest data from Flume. If the executor running the receiver fails, it currently has to be restarted on the same node to be able to receive data. This commit adds a new Sink which can be deployed to a Flume agent. This sink can be polled by a new DStream that is also included in this commit. This model ensures that data can be pulled into Spark from Flume even if the receiver is restarted on a new node. This also allows the receiver to receive data on multiple threads for better performance.
1 parent fed6303 commit 6d6776a

File tree

12 files changed

+885
-41
lines changed

12 files changed

+885
-41
lines changed

‎external/flume-sink/pom.xml‎

Lines changed: 82 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,82 @@
1+
<?xml version="1.0" encoding="UTF-8"?>
2+
<!--
3+
~ Licensed to the Apache Software Foundation (ASF) under one or more
4+
~ contributor license agreements. See the NOTICE file distributed with
5+
~ this work for additional information regarding copyright ownership.
6+
~ The ASF licenses this file to You under the Apache License, Version 2.0
7+
~ (the "License"); you may not use this file except in compliance with
8+
~ the License. You may obtain a copy of the License at
9+
~
10+
~ http://www.apache.org/licenses/LICENSE-2.0
11+
~
12+
~ Unless required by applicable law or agreed to in writing, software
13+
~ distributed under the License is distributed on an "AS IS" BASIS,
14+
~ WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
15+
~ See the License for the specific language governing permissions and
16+
~ limitations under the License.
17+
-->
18+
19+
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
20+
<modelVersion>4.0.0</modelVersion>
21+
<parent>
22+
<groupId>org.apache.spark</groupId>
23+
<artifactId>spark-parent</artifactId>
24+
<version>1.0.0-SNAPSHOT</version>
25+
<relativePath>../../pom.xml</relativePath>
26+
</parent>
27+
28+
<artifactId>spark-streaming-flume-sink_2.10</artifactId>
29+
<packaging>jar</packaging>
30+
<name>Spark Project External Flume Sink</name>
31+
<url>http://spark.apache.org/</url>
32+
<dependencies>
33+
<dependency>
34+
<groupId>org.apache.flume</groupId>
35+
<artifactId>flume-ng-sdk</artifactId>
36+
<version>1.4.0</version>
37+
<exclusions>
38+
<exclusion>
39+
<groupId>org.jboss.netty</groupId>
40+
<artifactId>netty</artifactId>
41+
</exclusion>
42+
<exclusion>
43+
<groupId>org.apache.thrift</groupId>
44+
<artifactId>libthrift</artifactId>
45+
</exclusion>
46+
</exclusions>
47+
</dependency>
48+
<dependency>
49+
<groupId>org.apache.flume</groupId>
50+
<artifactId>flume-ng-core</artifactId>
51+
<version>1.4.0</version>
52+
</dependency>
53+
</dependencies>
54+
<build>
55+
<outputDirectory>target/scala-${scala.binary.version}/classes</outputDirectory>
56+
<testOutputDirectory>target/scala-${scala.binary.version}/test-classes</testOutputDirectory>
57+
<plugins>
58+
<plugin>
59+
<groupId>org.scalatest</groupId>
60+
<artifactId>scalatest-maven-plugin</artifactId>
61+
</plugin>
62+
<plugin>
63+
<groupId>org.apache.avro</groupId>
64+
<artifactId>avro-maven-plugin</artifactId>
65+
<version>1.7.3</version>
66+
<configuration>
67+
<stringType>String</stringType>
68+
<!-- Generate the output in the same directory as the sbt-avro-plugin -->
69+
<outputDirectory>${project.basedir}/target/scala-${scala.binary.version}/src_managed/main/compiled_avro</outputDirectory>
70+
</configuration>
71+
<executions>
72+
<execution>
73+
<phase>generate-sources</phase>
74+
<goals>
75+
<goal>idl-protocol</goal>
76+
</goals>
77+
</execution>
78+
</executions>
79+
</plugin>
80+
</plugins>
81+
</build>
82+
</project>

‎external/flume-sink/src/main/avro/sparkflume.avdl‎

Lines changed: 40 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,40 @@
1+
/**
2+
* Licensed to the Apache Software Foundation (ASF) under one
3+
* or more contributor license agreements. See the NOTICE file
4+
* distributed with this work for additional information
5+
* regarding copyright ownership. The ASF licenses this file
6+
* to you under the Apache License, Version 2.0 (the
7+
* "License"); you may not use this file except in compliance
8+
* with the License. You may obtain a copy of the License at
9+
*
10+
* http://www.apache.org/licenses/LICENSE-2.0
11+
*
12+
* Unless required by applicable law or agreed to in writing,
13+
* software distributed under the License is distributed on an
14+
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
15+
* KIND, either express or implied. See the License for the
16+
* specific language governing permissions and limitations
17+
* under the License.
18+
*/
19+
20+
@namespace("org.apache.spark.flume")
21+
22+
protocol SparkFlumeProtocol {
23+
24+
record SparkSinkEvent {
25+
map<string> headers;
26+
bytes body;
27+
}
28+
29+
record EventBatch {
30+
string sequenceNumber;
31+
array<SparkSinkEvent> eventBatch;
32+
}
33+
34+
EventBatch getEventBatch (int n);
35+
36+
void ack (string sequenceNumber);
37+
38+
void nack (string sequenceNumber);
39+
40+
}

0 commit comments

Comments
 (0)