[SPARK-8378][Streaming]Add the Python API for Flume #6830
There are no files selected for viewing
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,55 @@ | ||
| # | ||
| # Licensed to the Apache Software Foundation (ASF) under one or more | ||
| # contributor license agreements. See the NOTICE file distributed with | ||
| # this work for additional information regarding copyright ownership. | ||
| # The ASF licenses this file to You 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. | ||
| # | ||
|
|
||
| """ | ||
| Counts words in UTF8 encoded, '\n' delimited text received from the network every second. | ||
| Usage: flume_wordcount.py <hostname> <port> | ||
|
|
||
| To run this on your local machine, you need to setup Flume first, see | ||
| https://flume.apache.org/documentation.html | ||
|
|
||
| and then run the example | ||
| `$ bin/spark-submit --jars external/flume-assembly/target/scala-*/\ | ||
| spark-streaming-flume-assembly-*.jar examples/src/main/python/streaming/flume_wordcount.py \ | ||
| localhost 12345 | ||
| """ | ||
| from __future__ import print_function | ||
|
|
||
| import sys | ||
|
|
||
| from pyspark import SparkContext | ||
| from pyspark.streaming import StreamingContext | ||
| from pyspark.streaming.flume import FlumeUtils | ||
|
|
||
| if __name__ == "__main__": | ||
| if len(sys.argv) != 3: | ||
| print("Usage: flume_wordcount.py <hostname> <port>", file=sys.stderr) | ||
| exit(-1) | ||
|
|
||
| sc = SparkContext(appName="PythonStreamingFlumeWordCount") | ||
| ssc = StreamingContext(sc, 1) | ||
|
|
||
| hostname, port = sys.argv[1:] | ||
| kvs = FlumeUtils.createStream(ssc, hostname, int(port)) | ||
| lines = kvs.map(lambda x: x[1]) | ||
| counts = lines.flatMap(lambda line: line.split(" ")) \ | ||
| .map(lambda word: (word, 1)) \ | ||
| .reduceByKey(lambda a, b: a+b) | ||
| counts.pprint() | ||
|
|
||
| ssc.start() | ||
| ssc.awaitTermination() |
| Original file line number | Diff line number | Diff line change |
|---|---|---|
| @@ -0,0 +1,135 @@ | ||
| <?xml version="1.0" encoding="UTF-8"?> | ||
| <!-- | ||
| ~ Licensed to the Apache Software Foundation (ASF) under one or more | ||
| ~ contributor license agreements. See the NOTICE file distributed with | ||
| ~ this work for additional information regarding copyright ownership. | ||
| ~ The ASF licenses this file to You 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 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"> | ||
| <modelVersion>4.0.0</modelVersion> | ||
| <parent> | ||
| <groupId>org.apache.spark</groupId> | ||
| <artifactId>spark-parent_2.10</artifactId> | ||
| <version>1.5.0-SNAPSHOT</version> | ||
| <relativePath>../../pom.xml</relativePath> | ||
| </parent> | ||
|
|
||
| <groupId>org.apache.spark</groupId> | ||
| <artifactId>spark-streaming-flume-assembly_2.10</artifactId> | ||
|
There was a problem hiding this comment. Hi guys, Sorry I'm late to the party, but why is this new assembly necessary? It creates an 80MB jar file that repackages a bunch of things already present in the Spark assembly (e.g. There was a problem hiding this comment. Nope, none of the fume stuff is present in the spark-assembly. That is On Mon, Jul 6, 2015 at 2:02 PM, Marcelo Vanzin notifications@github.com
There was a problem hiding this comment. Well, two things:
So any way you look at it, there is still something to be fixed here. There was a problem hiding this comment.
On Mon, Jul 6, 2015 at 2:19 PM, Marcelo Vanzin notifications@github.com
|
||
| <packaging>jar</packaging> | ||
| <name>Spark Project External Flume Assembly</name> | ||
| <url>http://spark.apache.org/</url> | ||
|
|
||
| <properties> | ||
| <sbt.project.name>streaming-flume-assembly</sbt.project.name> | ||
| </properties> | ||
|
|
||
| <dependencies> | ||
| <dependency> | ||
| <groupId>org.apache.spark</groupId> | ||
| <artifactId>spark-streaming-flume_${scala.binary.version}</artifactId> | ||
| <version>${project.version}</version> | ||
| </dependency> | ||
| <dependency> | ||
| <groupId>org.apache.spark</groupId> | ||
| <artifactId>spark-streaming_${scala.binary.version}</artifactId> | ||
| <version>${project.version}</version> | ||
| <scope>provided</scope> | ||
| </dependency> | ||
| <dependency> | ||
| <groupId>org.apache.avro</groupId> | ||
| <artifactId>avro</artifactId> | ||
| <version>${avro.version}</version> | ||
|
There was a problem hiding this comment. The dependencies of There was a problem hiding this comment. This seems like an avro bug. Can you file a jira for Avro? Avro should be compatible within the same minor version - 1.7.x There was a problem hiding this comment. Do you know where the other version of avro is coming up ? There was a problem hiding this comment.
I think
Actually "mvn dependency:tree" shows both |
||
| </dependency> | ||
| <dependency> | ||
| <groupId>org.apache.avro</groupId> | ||
| <artifactId>avro-ipc</artifactId> | ||
| <version>${avro.version}</version> | ||
| <exclusions> | ||
| <exclusion> | ||
| <groupId>io.netty</groupId> | ||
| <artifactId>netty</artifactId> | ||
| </exclusion> | ||
| <exclusion> | ||
| <groupId>org.mortbay.jetty</groupId> | ||
| <artifactId>jetty</artifactId> | ||
| </exclusion> | ||
| <exclusion> | ||
| <groupId>org.mortbay.jetty</groupId> | ||
| <artifactId>jetty-util</artifactId> | ||
| </exclusion> | ||
| <exclusion> | ||
| <groupId>org.mortbay.jetty</groupId> | ||
| <artifactId>servlet-api</artifactId> | ||
| </exclusion> | ||
| <exclusion> | ||
| <groupId>org.apache.velocity</groupId> | ||
|
There was a problem hiding this comment. Why do you need to pull all these in? Doesn't the spark-streaming-flume artifact pull these in anyway? There was a problem hiding this comment. These are exclusions.. not pulling in. There was a problem hiding this comment. ah, guess I need more coffee |
||
| <artifactId>velocity</artifactId> | ||
| </exclusion> | ||
| </exclusions> | ||
| </dependency> | ||
| </dependencies> | ||
|
|
||
| <build> | ||
| <outputDirectory>target/scala-${scala.binary.version}/classes</outputDirectory> | ||
| <testOutputDirectory>target/scala-${scala.binary.version}/test-classes</testOutputDirectory> | ||
| <plugins> | ||
| <plugin> | ||
| <groupId>org.apache.maven.plugins</groupId> | ||
| <artifactId>maven-shade-plugin</artifactId> | ||
| <configuration> | ||
| <shadedArtifactAttached>false</shadedArtifactAttached> | ||
| <outputFile>${project.build.directory}/scala-${scala.binary.version}/spark-streaming-flume-assembly-${project.version}.jar</outputFile> | ||
| <artifactSet> | ||
| <includes> | ||
| <include>*:*</include> | ||
| </includes> | ||
| </artifactSet> | ||
| <filters> | ||
| <filter> | ||
| <artifact>*:*</artifact> | ||
| <excludes> | ||
| <exclude>META-INF/*.SF</exclude> | ||
| <exclude>META-INF/*.DSA</exclude> | ||
| <exclude>META-INF/*.RSA</exclude> | ||
| </excludes> | ||
| </filter> | ||
| </filters> | ||
| </configuration> | ||
| <executions> | ||
| <execution> | ||
| <phase>package</phase> | ||
| <goals> | ||
| <goal>shade</goal> | ||
| </goals> | ||
| <configuration> | ||
| <transformers> | ||
| <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/> | ||
| <transformer implementation="org.apache.maven.plugins.shade.resource.AppendingTransformer"> | ||
| <resource>reference.conf</resource> | ||
| </transformer> | ||
| <transformer implementation="org.apache.maven.plugins.shade.resource.DontIncludeResourceTransformer"> | ||
| <resource>log4j.properties</resource> | ||
| </transformer> | ||
| <transformer implementation="org.apache.maven.plugins.shade.resource.ApacheLicenseResourceTransformer"/> | ||
| <transformer implementation="org.apache.maven.plugins.shade.resource.ApacheNoticeResourceTransformer"/> | ||
| </transformers> | ||
| </configuration> | ||
| </execution> | ||
| </executions> | ||
| </plugin> | ||
| </plugins> | ||
| </build> | ||
| </project> | ||
|
|
||
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
Is the decoding logic the same here too? UTF-8 encoded string, or custom decoding function? If yes, we should move that snippet explaining this outside of both approaches and specify that it is applicable to both.
There was a problem hiding this comment.
Choose a reason for hiding this comment
The reason will be displayed to describe this comment to others. Learn more.
I updated the doc. Because there is no a Python example for
FlumeUtils.createPollingStream(it requires the user installing the Spark sink jar to Flume, so we cannot write an out-of-the-box example), I just copied the description of the encoding function to here.