<div data-lang="python" markdown="1">

from pyspark.streaming.flume import FlumeUtils

flumeStream = FlumeUtils.createStream(streamingContext, [chosen machine's hostname], [chosen port])

By default, the Python API will decode Flume event body as UTF8 encoded strings. You can specify your custom decoding function to decode the body byte arrays in Flume events to any arbitrary data type.

See the [API docs](api/python/pyspark.streaming.html#pyspark.streaming.flume.FlumeUtils)

and the [example]({{site.SPARK_GITHUB_URL}}/blob/master/examples/src/main/python/streaming/flume_wordcount.py).

</div>

<div data-lang="python" markdown="1">

from pyspark.streaming.flume import FlumeUtils

addresses = [([sink machine hostname 1], [sink port 1]), ([sink machine hostname 2], [sink port 2])]

flumeStream = FlumeUtils.createPollingStream(streamingContext, addresses)

</div>

out of these sources, *only* Kafka and Flume are available in the Python API. We will add more advanced sources in the Python API in future.

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()

<?xml version="1.0" encoding="UTF-8"?>

<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>

<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>

</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>

<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>

<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>

import java.io.{DataOutputStream, ByteArrayOutputStream}

import java.util.{List => JList, Map => JMap}

import scala.collection.JavaConversions._

import org.apache.spark.api.java.function.PairFunction

import org.apache.spark.api.python.PythonRDD

import org.apache.spark.streaming.api.java.{JavaPairDStream, JavaReceiverInputDStream, JavaStreamingContext}

private class FlumeUtilsPythonHelper {

def createStream(

jssc: JavaStreamingContext,

hostname: String,

port: Int,

storageLevel: StorageLevel,

enableDecompression: Boolean

): JavaPairDStream[Array[Byte], Array[Byte]] = {

val dstream = FlumeUtils.createStream(jssc, hostname, port, storageLevel, enableDecompression)

FlumeUtilsPythonHelper.toDStreamForPython(dstream)

}

def createPollingStream(

jssc: JavaStreamingContext,

hosts: JList[String],

ports: JList[Int],

storageLevel: StorageLevel,

maxBatchSize: Int,

parallelism: Int

): JavaPairDStream[Array[Byte], Array[Byte]] = {

assert(hosts.length == ports.length)

val addresses = hosts.zip(ports).map {

case (host, port) => new InetSocketAddress(host, port)

}

val dstream = FlumeUtils.createPollingStream(

jssc.ssc, addresses, storageLevel, maxBatchSize, parallelism)

FlumeUtilsPythonHelper.toDStreamForPython(dstream)

}

}

private object FlumeUtilsPythonHelper {

private def stringMapToByteArray(map: JMap[CharSequence, CharSequence]): Array[Byte] = {

val byteStream = new ByteArrayOutputStream()

val output = new DataOutputStream(byteStream)

try {

output.writeInt(map.size)

map.foreach { kv =>

PythonRDD.writeUTF(kv._1.toString, output)

PythonRDD.writeUTF(kv._2.toString, output)

}

byteStream.toByteArray

}

finally {

output.close()

}

}

private def toDStreamForPython(dstream: JavaReceiverInputDStream[SparkFlumeEvent]):

JavaPairDStream[Array[Byte], Array[Byte]] = {

dstream.mapToPair(new PairFunction[SparkFlumeEvent, Array[Byte], Array[Byte]] {

override def call(sparkEvent: SparkFlumeEvent): (Array[Byte], Array[Byte]) = {

val event = sparkEvent.event

val byteBuffer = event.getBody

val body = new Array[Byte](byteBuffer.remaining())

byteBuffer.get(body)

(stringMapToByteArray(event.getHeaders), body)

}

})

}

}

<module>external/flume-assembly</module>

val assemblyProjects@Seq(assembly, examples, networkYarn, streamingFlumeAssembly, streamingKafkaAssembly) =

Seq("assembly", "examples", "network-yarn", "streaming-flume-assembly", "streaming-kafka-assembly")

if (mName.contains("streaming-flume-assembly") || mName.contains("streaming-kafka-assembly")) {