Skip to content

[SPARK-8564][Streaming]Add the Python API for Kinesis #6955

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.

Sign up for GitHub

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

Closed
zsxwing wants to merge 16 commits into from
Closed

[SPARK-8564][Streaming]Add the Python API for Kinesis #6955

Show file tree
Hide file tree
Changes from 1 commit
Commits
Show all changes
16 commits
Select commit Hold shift + click to select a range
4957c0b
Add the Python API for Kinesis
zsxwing Jun 23, 2015
add2beb
Merge branch 'master' into kinesis-python
zsxwing Jul 22, 2015
687446b
Fix the error message and the maven output path
zsxwing Jul 23, 2015
3da2601
Fix the kinesis folder
zsxwing Jul 23, 2015
e33d505
Merge remote-tracking branch 'origin/master' into kinesis-python
zsxwing Jul 23, 2015
466b425
Add python tests for Kinesis
zsxwing Jul 23, 2015
cc9d071
Fix the python test errors
zsxwing Jul 23, 2015
7c5cfb0
RUN_KINESIS_TESTS -> ENABLE_KINESIS_TESTS
zsxwing Jul 23, 2015
6c37395
Print stack trace for debug
zsxwing Jul 24, 2015
756a128
Merge branch 'master' into kinesis-python
zsxwing Jul 24, 2015
96670ff
Fix the compilation error after merging master
zsxwing Jul 24, 2015
fca416b
Fix wrong comparison
zsxwing Jul 24, 2015
5082d28
Fix the syntax error for Python 2.6
zsxwing Jul 26, 2015
32e6451
Merge remote-tracking branch 'origin/master' into kinesis-python
zsxwing Jul 30, 2015
455f7ea
Remove streaming_kinesis_asl_assembly module and simply add the sourc…
zsxwing Jul 30, 2015
e42e471
Merge branch 'master' into kinesis-python
zsxwing Jul 31, 2015
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
Prev Previous commit
Next Next commit
Merge branch 'master' into kinesis-python
  • Loading branch information
@zsxwing
zsxwing committed Jul 22, 2015
commit add2bebd2b97c0610882928ac9f5893bbeff36cb
287 changes: 1 addition & 286 deletions dev/run-tests.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -35,292 +35,6 @@
# -------------------------------------------------------------------------------------------------


all_modules = []


class Module(object):
"""
A module is the basic abstraction in our test runner script. Each module consists of a set of
source files, a set of test commands, and a set of dependencies on other modules. We use modules
to define a dependency graph that lets determine which tests to run based on which files have
changed.
"""

def __init__(self, name, dependencies, source_file_regexes, build_profile_flags=(),
sbt_test_goals=(), should_run_python_tests=False, should_run_r_tests=False):
"""
Define a new module.

:param name: A short module name, for display in logging and error messages.
:param dependencies: A set of dependencies for this module. This should only include direct
dependencies; transitive dependencies are resolved automatically.
:param source_file_regexes: a set of regexes that match source files belonging to this
module. These regexes are applied by attempting to match at the beginning of the
filename strings.
:param build_profile_flags: A set of profile flags that should be passed to Maven or SBT in
order to build and test this module (e.g. '-PprofileName').
:param sbt_test_goals: A set of SBT test goals for testing this module.
:param should_run_python_tests: If true, changes in this module will trigger Python tests.
For now, this has the effect of causing _all_ Python tests to be run, although in the
future this should be changed to run only a subset of the Python tests that depend
on this module.
:param should_run_r_tests: If true, changes in this module will trigger all R tests.
"""
self.name = name
self.dependencies = dependencies
self.source_file_prefixes = source_file_regexes
self.sbt_test_goals = sbt_test_goals
self.build_profile_flags = build_profile_flags
self.should_run_python_tests = should_run_python_tests
self.should_run_r_tests = should_run_r_tests

self.dependent_modules = set()
for dep in dependencies:
dep.dependent_modules.add(self)
all_modules.append(self)

def contains_file(self, filename):
return any(re.match(p, filename) for p in self.source_file_prefixes)


sql = Module(
name="sql",
dependencies=[],
source_file_regexes=[
"sql/(?!hive-thriftserver)",
"bin/spark-sql",
],
build_profile_flags=[
"-Phive",
],
sbt_test_goals=[
"catalyst/test",
"sql/test",
"hive/test",
]
)


hive_thriftserver = Module(
name="hive-thriftserver",
dependencies=[sql],
source_file_regexes=[
"sql/hive-thriftserver",
"sbin/start-thriftserver.sh",
],
build_profile_flags=[
"-Phive-thriftserver",
],
sbt_test_goals=[
"hive-thriftserver/test",
]
)


graphx = Module(
name="graphx",
dependencies=[],
source_file_regexes=[
"graphx/",
],
sbt_test_goals=[
"graphx/test"
]
)


streaming = Module(
name="streaming",
dependencies=[],
source_file_regexes=[
"streaming",
],
sbt_test_goals=[
"streaming/test",
]
)


streaming_kinesis_asl = Module(
name="kinesis-asl",
dependencies=[streaming],
source_file_regexes=[
"extras/kinesis-asl/",
],
build_profile_flags=[
"-Pkinesis-asl",
],
sbt_test_goals=[
"kinesis-asl/test",
]
)


streaming_kinesis_asl_assembly = Module(
name="kinesis-asl-assembly",
dependencies=[streaming_kinesis_asl],
source_file_regexes=[
"extras/kinesis-asl-assembly/",
],
build_profile_flags=[
"-Pkinesis-asl",
]
)


streaming_zeromq = Module(
name="streaming-zeromq",
dependencies=[streaming],
source_file_regexes=[
"external/zeromq",
],
sbt_test_goals=[
"streaming-zeromq/test",
]
)


streaming_twitter = Module(
name="streaming-twitter",
dependencies=[streaming],
source_file_regexes=[
"external/twitter",
],
sbt_test_goals=[
"streaming-twitter/test",
]
)


streaming_mqtt = Module(
name="streaming-mqtt",
dependencies=[streaming],
source_file_regexes=[
"external/mqtt",
],
sbt_test_goals=[
"streaming-mqtt/test",
]
)


streaming_kafka = Module(
name="streaming-kafka",
dependencies=[streaming],
source_file_regexes=[
"external/kafka",
"external/kafka-assembly",
],
sbt_test_goals=[
"streaming-kafka/test",
]
)


streaming_flume_sink = Module(
name="streaming-flume-sink",
dependencies=[streaming],
source_file_regexes=[
"external/flume-sink",
],
sbt_test_goals=[
"streaming-flume-sink/test",
]
)


streaming_flume = Module(
name="streaming_flume",
dependencies=[streaming],
source_file_regexes=[
"external/flume",
],
sbt_test_goals=[
"streaming-flume/test",
]
)


mllib = Module(
name="mllib",
dependencies=[streaming, sql],
source_file_regexes=[
"data/mllib/",
"mllib/",
],
sbt_test_goals=[
"mllib/test",
]
)


examples = Module(
name="examples",
dependencies=[graphx, mllib, streaming, sql],
source_file_regexes=[
"examples/",
],
sbt_test_goals=[
"examples/test",
]
)


pyspark = Module(
name="pyspark",
dependencies=[mllib, streaming, streaming_kafka, streaming_kinesis_asl_assembly, sql],
source_file_regexes=[
"python/"
],
should_run_python_tests=True
)


sparkr = Module(
name="sparkr",
dependencies=[sql, mllib],
source_file_regexes=[
"R/",
],
should_run_r_tests=True
)


docs = Module(
name="docs",
dependencies=[],
source_file_regexes=[
"docs/",
]
)


ec2 = Module(
name="ec2",
dependencies=[],
source_file_regexes=[
"ec2/",
]
)


# The root module is a dummy module which is used to run all of the tests.
# No other modules should directly depend on this module.
root = Module(
name="root",
dependencies=[],
source_file_regexes=[],
# In order to run all of the tests, enable every test profile:
build_profile_flags=
list(set(itertools.chain.from_iterable(m.build_profile_flags for m in all_modules))),
sbt_test_goals=[
"test",
],
should_run_python_tests=True,
should_run_r_tests=True
)


def determine_modules_for_files(filenames):
"""
Given a list of filenames, return the set of modules that contain those files.
Expand Down Expand Up @@ -580,6 +294,7 @@ def build_spark_sbt(hadoop_version):
sbt_goals = ["package",
"assembly/assembly",
"streaming-kafka-assembly/assembly",
"streaming-flume-assembly/assembly",
"streaming-kinesis-asl-assembly/assembly"]
profiles_and_goals = build_profiles + sbt_goals

Expand Down
20 changes: 19 additions & 1 deletion dev/sparktestsupport/modules.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -141,6 +141,18 @@ def contains_file(self, filename):
)


streaming_kinesis_asl_assembly = Module(
name="kinesis-asl-assembly",
dependencies=[streaming_kinesis_asl],
source_file_regexes=[
"extras/kinesis-asl-assembly/",
],
build_profile_flags=[
"-Pkinesis-asl",
]
)


streaming_zeromq = Module(
name="streaming-zeromq",
dependencies=[streaming],
Expand Down Expand Up @@ -290,7 +302,13 @@ def contains_file(self, filename):

pyspark_streaming = Module(
name="pyspark-streaming",
dependencies=[pyspark_core, streaming, streaming_kafka, streaming_flume_assembly],
dependencies=[
pyspark_core,
streaming,
streaming_kafka,
streaming_flume_assembly,
streaming_kinesis_asl_assembly
],
source_file_regexes=[
"python/pyspark/streaming"
],
Expand Down
6 changes: 3 additions & 3 deletions project/SparkBuild.scala
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -45,8 +45,8 @@ object BuildCommons {
sparkKinesisAsl) = Seq("yarn", "yarn-stable", "java8-tests", "ganglia-lgpl",
"kinesis-asl").map(ProjectRef(buildLocation, _))

val assemblyProjects@Seq(assembly, examples, networkYarn, streamingKafkaAssembly, streamingKinesisAslAssembly) =
Seq("assembly", "examples", "network-yarn", "streaming-kafka-assembly", "streaming-kinesis-asl-assembly")
val assemblyProjects@Seq(assembly, examples, networkYarn, streamingFlumeAssembly, streamingKafkaAssembly, streamingKinesisAslAssembly) =
Seq("assembly", "examples", "network-yarn", "streaming-flume-assembly", "streaming-kafka-assembly", "streaming-kinesis-asl-assembly")
.map(ProjectRef(buildLocation, _))

val tools = ProjectRef(buildLocation, "tools")
Expand Down Expand Up @@ -351,7 +351,7 @@ object Assembly {
.getOrElse(SbtPomKeys.effectivePom.value.getProperties.get("hadoop.version").asInstanceOf[String])
},
jarName in assembly <<= (version, moduleName, hadoopVersion) map { (v, mName, hv) =>
if (mName.contains("streaming-kafka-assembly" | "streaming-kinesis-asl-assembly")) {
if (mName.contains("streaming-flume-assembly") || mName.contains("streaming-kafka-assembly") || mName.contains("streaming-kinesis-asl-assembly")) {
// This must match the same name used in maven (see external/kafka-assembly/pom.xml)
s"${mName}-${v}.jar"
} else {
Expand Down
Loading
You are viewing a condensed version of this merge commit. You can view the full changes here.