[SPARK-7481] [build] Add spark-cloud module to pull in object store access, + documentation

[steveloughran]

What changes were proposed in this pull request?

Add a new cloud module and maven profile to pull in object store support from hadoop-openstack, hadoop-aws and hadoop-azure JARs, along with their dependencies.

As a result, it restores s3n:// access to S3, adds its s3a:// replacement , OpenStack swift:// and azure wasb://.

The patch defines a new module, spark-cloud with transitive dependencies on the amazon and azure JARs. The spark assembly module JAR depends on this module to pull the relevant hadoop and dependent JARs into SPARK_HOME/jars. This module is only included in a build if the -cloud profile is set; there's an implicit dependency on the hadoop-2.7 module too, because that is where the various hadoop-$CLOUD modules are imported.

There's a documentation page, cloud_integration.md, which covers the details of using Spark with object stores: API level, the different object stores, limitations, recommended tuning options. All the examples are in Scala. The docs try to be an introduction which avoids cut-and-paste text which will age fast against different versions, pointing at the Hadoop docs.

Having an explicit module enables downstream applications to declare their dependency upon spark-cloud and get object store access. This is used for the integration tests, which therefore implicitly test the correctness of the declared dependencies.

There is no import or coverage of the google cloud storage gs:// storage; that's not an ASF project & we've actually had some problems with gs; at the time of writing it hasn't implemented the standard Hadoop FS contract tests, so it's not clear what state is in. That could be future work.

There is a mention but no details on Amazon EMR's S3:// client. It's essentially says "Amazon's code, their problem", which is the case.

How was this patch tested?

There are no functional tests in this module; it is left to downstream projects to implement integration tests. I have such tests in https://github.com/steveloughran/spark-cloud-examples/tree/master/cloud-examples

The module has been tested against: Hadoop 2.7.3, branch-2.8 (2.8.0-SNAPSHOT), branch-2 (2.9.0-SNAPSHOT) and Hadoop trunk;

Tested endpoints: AWS S3 US-East, AWS S3 Ireland, Azure (US and EU), and Rackspace Swift US and EU,

[steveloughran]

Note that as this patch is is playing with the maven build and the hadoop-2.6 and hadoop-2.7 profiles, the SparkQA builds aren't going to pick up on much here.

[nchammas]

Typo: By the sbt build.

[steveloughran]

funny: that was a comment I lifted with the dependency cargo cult style; it'll need fixing in in the original too..

[SparkQA]

Test build #54333 has finished for PR 12004 at commit 5e9cfbe.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[steveloughran]

test failures are in hive; unrelated

[SparkQA]

Test build #54454 has finished for PR 12004 at commit 72b3548.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds the following public classes (experimental):
  • class DependencyCheckSuite extends SparkFunSuite

[steveloughran]

build failing as SBT needs to be conditional on the spark/cloud module being Hadoop 2.6+

[SparkQA]

Test build #54525 has finished for PR 12004 at commit 6beafb5.

• This patch fails MiMa tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #54803 has finished for PR 12004 at commit 6487d93.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #55916 has finished for PR 12004 at commit 105de0b.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #56092 has finished for PR 12004 at commit ce48b8c.

• This patch fails Spark unit tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #56360 has finished for PR 12004 at commit 8845af0.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #56390 has finished for PR 12004 at commit 2fca815.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[steveloughran]

The latest version of this does, among other things, call FileSystem.toString after operations. In HADOOP-13028, along with seek optimisation, S3aFileSystem.toString() now dumps all the statistics to date. This means that the aggregate state of all test runs are displayed; if you run a specific test standalone you can see the stats purely for that test

Here's a test with the maven args -Phadoop-2.7 -DwildcardSuites=org.apache.spark.cloud.s3.S3aIOSuite -Dcloud.test.configuration.file=../cloud.xml -Dhadoop.version=2.9.0-SNAPSHOT -Dtest.method.keys=CSVgz

2016-04-26 14:32:17,104 INFO  scheduler.TaskSetManager (Logging.scala:logInfo(54)) - Starting task 0.0 in stage 0.0 (TID 0, localhost, partition 0,PROCESS_LOCAL, 5261 bytes)
2016-04-26 14:32:17,105 INFO  executor.Executor (Logging.scala:logInfo(54)) - Running task 0.0 in stage 0.0 (TID 0)
2016-04-26 14:32:17,111 INFO  rdd.HadoopRDD (Logging.scala:logInfo(54)) - Input split: s3a://landsat-pds/scene_list.gz:0+20430493
2016-04-26 14:32:17,285 INFO  compress.CodecPool (CodecPool.java:getDecompressor(181)) - Got brand-new decompressor [.gz]
2016-04-26 14:32:21,724 INFO  executor.Executor (Logging.scala:logInfo(54)) - Finished task 0.0 in stage 0.0 (TID 0). 2643 bytes result sent to driver
2016-04-26 14:32:21,727 INFO  scheduler.TaskSetManager (Logging.scala:logInfo(54)) - Finished task 0.0 in stage 0.0 (TID 0) in 4625 ms on localhost (1/1)
2016-04-26 14:32:21,727 INFO  scheduler.TaskSchedulerImpl (Logging.scala:logInfo(54)) - Removed TaskSet 0.0, whose tasks have all completed, from pool 
2016-04-26 14:32:21,728 INFO  scheduler.DAGScheduler (Logging.scala:logInfo(54)) - ResultStage 0 (count at S3aIOSuite.scala:127) finished in 4.626 s
2016-04-26 14:32:21,728 INFO  scheduler.DAGScheduler (Logging.scala:logInfo(54)) - Job 0 finished: count at S3aIOSuite.scala:127, took 4.636417 s
2016-04-26 14:32:21,729 INFO  s3.S3aIOSuite (Logging.scala:logInfo(54)) -  size of s3a://landsat-pds/scene_list.gz = 464105 rows read in 4815885000 nS
2016-04-26 14:32:21,729 INFO  s3.S3aIOSuite (Logging.scala:logInfo(54)) - Filesystem statistics S3AFileSystem{uri=s3a://landsat-pds, workingDir=s3a://landsat-pds/user/stevel, partSize=104857600, enableMultiObjectsDelete=true, multiPartThreshold=2147483647, statistics {40864879 bytes read, 7786 bytes written, 110 read ops, 0 large read ops, 26 write ops}, metrics {{Context=S3AFileSystem} {FileSystemId=bc5db77d-e17d-41bb-88ab-44b26cf3eda4-landsat-pds} {fsURI=s3a://landsat-pds/scene_list.gz} {files_created=0} {files_copied=0} {files_copied_bytes=0} {files_deleted=0} {directories_created=0} {directories_deleted=0} {ignored_errors=0} {streamForwardSeekOperations=0} {streamCloseOperations=2} {streamBytesSkippedOnSeek=0} {streamReadOperations=2821} {streamReadExceptions=0} {streamAborted=0} {streamBackwardSeekOperations=0} {streamClosed=2} {streamOpened=2} {streamSeekOperations=0} {streamBytesRead=40860986} {streamReadOperationsIncomplete=2821} {streamReadFullyOperations=0} }}
2016-04-26 14:32:21,729 INFO  s3.S3aIOSuite (Logging.scala:logInfo(54)) - 

[steveloughran]

Oh, and there's an initial documentation page on spark + cloud infrastructure, which tries to make clear that object stores are not real filesystems

[SparkQA]

Test build #56998 has finished for PR 12004 at commit 8926acb.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[steveloughran]

Note that as this module only builds on Hadoop >= 2.6; jenkins won't be compiling it. The tests are designed to skip running if no config file to cloud infrastructure has been provided.

[SparkQA]

Test build #57003 has finished for PR 12004 at commit 4e4e941.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[SparkQA]

Test build #57782 has finished for PR 12004 at commit 70065dc.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[steveloughran]

For anyone trying to run these tests, they'll need a test xml file and refer to it

mvn test -Phadoop-2.6 -Dcloud.test.configuration.file=../cloud.xml 

The referenced file uses XInclude to input the AWS credentials which I keep a long way away from SCM-managed directories.

<?xml version="1.0"?>
<?xml-stylesheet type="text/xsl" href="configuration.xsl"?>
<!--
  ~ 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.
  -->

<configuration>
  <include xmlns="http://www.w3.org/2001/XInclude"
    href="file:///home/stevel/.aws/keys.xml"/>

  <property>
    <name>aws.tests.enabled</name>
    <value>true</value>
  </property>

  <property>
    <name>s3a.test.uri</name>
    <value>s3a://test-eu1</value>
  </property>
</configuration>

All the test suites will be designed to run iff the relevant enabled.flag is set; this is why there's a new method to declare tests, ctest(key: String, summary: String, detail: String)(testFun: => Unit): Unit

these tests are not only conditional on the suite being enabled, they each have a key which can be explicitly named from the build in the test.method.keys attr. This allows explicit methods to be named the way the current maven surefire runner doesn't; the time it can take to run individual tests makes this feature invaluable during iterative development.

  ctest("CSVgz", "Read compressed CSV",
    "Read compressed CSV files through the spark context") {
    val source = SceneList
    sc = new SparkContext("local", "test", newSparkConf(source))
    val sceneInfo = getFS(source).getFileStatus(source)
    logInfo(s"Compressed size = ${sceneInfo.getLen}")
    validateCSV(sc, source)
    logInfo(s"Filesystem statistics ${getFS(source)}")
  }

For best performance, you need a build of hadoop which has as much of HADOOP-11694 applied, especially HADOOP-12444, lazy seek (in branch-2.8 already), and HADOOP-13028

mvn test -Phadoop-2.7 -DwildcardSuites=org.apache.spark.cloud.s3.S3aIOSuite  \
-Dcloud.test.configuration.file=../cloud.xml \
-Dhadoop.version=2.9.0-SNAPSHOT \
-Dtest.method.keys=CSVgz

[SparkQA]

Test build #58856 has finished for PR 12004 at commit 4e37c7a.

• This patch fails Scala style tests.
• This patch merges cleanly.
• This patch adds no public classes.

[steveloughran]

The latest patch embraces the fact that 2.6 is the base hadoop version so the hadoop-aws JAR is always pulled in, dependencies set up. One thing to bear in mind here that the [Phase I fixes|https://issues.apache.org/jira/browse/HADOOP-11571] aren't in there, And s3a absolutely must not be used in production, the big killers being:

• HADOOP-11570 closing the stream reads to the EOF, which means every seek() can read in 2x file size.
• HADOOP-11584 block size returned in getFileStatus() ==0. That is bad because both Pig and Spark use that block size in partitioning, so will split up a file into single byte partitions: 20MB file, 210^7 tasks. Each of which will open the file at byte (0), then call seek to offset, then close(). As a result, 210e7 * tasks reading 2* 2 2 * 10e7 bytes. This is generally considered "pathologically suboptimal". I've had to modify my downstream tests to recognise when the block size of a file ==0 and skip those tests.

s3n will work; in 2.6 it moved to the aws JAR, so reinstate the functionality which was in spark builds against hadoop 2.2-2.5

Between this patch and the previous one, I had to try to switch spark-hadoop-cloud to move to a POM-only artifact. Doesn't work for downstream projects: they kept getting the 2.6.5 artifacts irrespective of which profile was used at build time. That negates the whole goal of the module "give consistent access to the hadoop cloud modules". This patch produces a minimal JAR for this reason

[SparkQA]

Test build #74899 has finished for PR 12004 at commit 83d9368.

• This patch passes all tests.
• This patch merges cleanly.
• This patch adds no public classes.

[steveloughran]

Any comments on the latest patch? Anyone?

[steveloughran]

@srowen anything else I need to do here?

[srowen]

Same, I've timed out on this and am not going to review it further. I don't think this is worth this much effort and discussion.

[srowen]

We may need to make this 2.3.0-SNAPSHOT now, because that's what's correct for master, then change it if it back-ports.

[steveloughran]

I'd noticed that this morning....

[srowen]

I would only include the first sentence here. The description here should be short since nobody will likely read it. Anything substantive could go in docs.

[srowen]

Call this hadoop-cloud perhaps?

[steveloughran]

so org/apache/spark + hadoop-cloud? I'll cause too much confusion were any JAR created thrown into a lib/ directory; you'd get

hadoop-aws-2.8.1.jar
spark-core-2.3.0
hadoop-cloud-2.3.0

& people would be trying to understand why the hadoop-* was out of sync, who to ping, etc.

There's actually a hadoop-cloudproject POM coming in hadoop-trunk to try and be a one-stop-dependency for all cloud bindings (avoiding the ongoing "declare new dependencies per version"). the names are way too close.

I'd had it as spark-cloud, you'd felt spark-hadoop-cloud was better. I can't think of what else would do, but I do think spark- is the string which should go at the front

[srowen]

So this only needs to come in for Hadoop 2.7+, not 2.6?

[steveloughran]

yes

• 2.7 adds hadoop-azure for wasb:
• 2.8 adds hadoop-azure-datalake for adl:

There's going to be an aggregate POM in trunk, hadoop-cloud-storage, which declares all transitive stuff, ideally stripping down cruft we don't need. That way if new things go in, anything pulling that JAR shouldn't have to add new declarations. There's still the problem of transitive breakage of JARs (i.e. Jackson)

[srowen]

Google is "Google Cloud Platform"? or maybe just say "All major public cloud infrastructure providers offer ..."

[steveloughran]

I'll do "All major cloud providers offer persistent data storage in object stores."

[srowen]

same comment about the dash here

[steveloughran]

again, cut that whole section. Left to the Hadoop docs

[srowen]

This is all true, but is it relevant to a Spark app? maybe, if it's also using HDFS APIs directly. If you revise this bit, consider emphasizing how this reality affects Spark usage in particular.

[steveloughran]

good point. I'll call that out up front

1. Reading and writing data can be slower than expected.
2. Some directory structures may be very inefficient to scan during query split calculation.
3. The output of saved RDD may not be immediately visible to a follow-on query.
4. The internal mechanism by which Spark commits work when saving an RDD is potentially
   both slow and unreliable.

That's essentially it on day to day use.

[srowen]

This section I'm less clear is relevant to a Spark app. Anything's possible but it's rare that someone would write a stream directly (?) Dunno, it's not bad info, just want to balance writing and maintaining this info in Spark docs vs pointing to other resources with a summary of key points to know.

[steveloughran]

I've cut it

[steveloughran]

(FWIW, where this does cause problems is that really slow writes can block things like heartbeating protocols if the same thread is doing the heartbeat. Hasn't surfaced in spark AFAIK.

[srowen]

what drives this -- is it just another Jackson component whose version has to be harmonized?

[steveloughran]

yes, keeping jackson in sync is a key breakage point. Declaring it in the root POM doesn't add it everywhere, it just delares it so that the cloud POM can exclude the one which comes via the aws-java-sdk-s3 dependency JARs and pick up the one used in spark. The (later) Spark one is compatible with the one aws SDK depends on, so moving up works...it's just that all jackson bits needs to be in sync, and there's no way in Maven or Ivy to declare that fact.

[srowen]

Shouldn't all this only be in cloud/pom.xml?

[steveloughran]

I'm declaring them here purely as declarations, so there's one place where all the hadoop dependencies go, versions declared etc. The cloud POM explicitly adds them to the module dependency, so that's the only place where they get pulled into the build. It's similar to what's done with Hive: spark-hive declares the actual dependencies, but the root pom declares what hive version to pull in, and what stuff to leave out.

[steveloughran]

github isn't letting me reopen this, so I'm going to submit the patch with reworked docs as a new PR. The machines do not like me today.