Isolation is not participation

First

1. I speak only for myself as an individual, not representing any current or previous employer, ASF ...etc.
2. I'm not going anywhere near business aspects; out of scope.
3. I am only looking at Apache Hadoop, which is of course the foundation of Amazon EMR. Which also means: if someone says "yes but projects X, Y & Z, ..." my response is "that's nice, but coming back to the project under discussion, ..."
4. lots of people I know & respect work for Amazon. I am very much looking at the actions of the company, not the individuals.
5. And I'm not making any suggestions about what people should do, only arguing that the current stance is harmful to everyone.
6. I saw last week that EMR now has a reimplementation of the S3A committers, without any credit whatsoever for something I consider profound. This means I'm probably a bit sensitive right now. I waited a couple of days before finishing this  post,

With that out the way:-

As I type this a nearby terminal window runs MiniMR jobs against a csv.gz file listing AWS landsat photos, stored somewhere in a bucket.

The tests run on a macbook, a distant descendant of BSD linux, Mach Kernel and its incomplete open source sibling, Darwin. Much of the dev tooling I use is all open source, downloaded via homebrew. The network connection is via a router running DD-WRT.


That Landsat file, s3a://landsat-pds/scene_list.gz, is arguably the single main contribution from the AWS infra for that Hadoop testing.

It's a few hundred MB of free to use data, so I've used it for IO seek/read performance tests, spark dataframe queries, and now, SQL statements direct to the storage infrastructure. Those test are also where I get to explore the new features of the java language, LambdaTestUtils, which is my lifting of what I consider to be the best bits of scalatest. Now I'm adding async IO operations to the Hadoop FileSystem/FileContext classes, and in the tests I'm learning about the java 8+ completable future stuff, how to get them to run IOException-raising code (TL;DR: it hurts)

While I wait for my tests to finish, I see, there's a lot of online discussion about could providers and open source projects, especially post AWS re:Invent (re:Package?), so I'd thought I'd join in.


Of all the bits of recent writing on the topic, one I really like is Roman's, which focuses a lot on community over code.

That is a key thing: open source development is a community. And members of that community can participate by

1. writing code
2. reviewing code
3. writing docs
4. reviewing docs
5. testing releases, nightly builds
6. helping other people with their problems
7. helping other projects who are having problems with your project's code.
8. helping other projects take ideas from your code and make use of it
9. filing bug reports
10. reviewing, commenting, on, maybe even fixing bug reports.
11. turning up a conferences, talking about what you are doing, sharing
12. listening. Not just to people who pay you money, but people who want to use the stuff you've written.
13. helping build that community by encouraging the participation of others, nurturing their contributions along, trying to get them to embrace your code and testing philosophy, etc.

There are more, but those are some of the key ones.

A key, recurrent theme is that community, where you can contribute in many ways, but you do have to be proactive to build that community. And the best ASF projects are ones which have a broad set of contributors

Take for example, the grand Java 9, 10, 11 project: [HADOOP-11123, HADOOP-11423, HADOOP-15338 ]. That's an epic of suffering, primarily taken on by Akira Ajisaka, and Takanobu Asanuma at Yahoo! Japan, and a few other brave people. This isn't some grand "shout about this at keynotes" kind of feature, but its a critical contribution by people who rely on Hadoop and have a pressing need "Run on Java 9+", and are prepared to put in the effort. I watch their JIRAs with awe.

That's a community effort, driven by users with needs.

Another interesting bit of work: Multipart Upload from HDFS to other block stores. I've  been the specification police there; my reaction to "adding fundamental APIs without strict specification and tests" was predictable, so I don't know why they tried to get it past me. Who did that work? Ewan Higgs at Western Digital did a lot -they can see real benefit for their enterprise object store. Virajith Jalaparti at Microsoft, People who want HDFS to be able to use their storage systems for the block store. And there's another side-effect: that mulitpart upload API essentially provides a standard API for multipart-upload based committers. For the S3A committers we added our own private API to S3A FS "WriteOperationHelper"; this will give it to every FS which supports it. And you can do a version of DistCP which writes blocks to the store in parallel from across the filesystem...a very high performance blockstore-to-block-or-object store copy mechanism.

This work is all being done by people who sell storage in one form or another, who see value in the feature, and are putting in the effort to develop it in the open, encourage participation from others, and deliver  something independent of underlying storage

This bit of work highlights something else: that Hadoop FS API is broadly used way beyond HDFS, and we need to evolve it to deal with things HDFS offers but keeps hidden, but also for object stores, whose characteristics involve:

• very expensive cost of seek(), especially given ORC and Parquet know their read plan way beyond the next read. Fix: HADOOP-11867: Vectorized Read Operations,
• very expensive cost of mimicking hierarchical directory trees, treewalking is not ideal,
• slow to open a file, as even the existence check can take hundreds of milliseconds.
• Interesting new failure modes. Throttling/503 responses if you put too much load on a shard of the store, for example. Which can surface anywhere.
• Variable rename performance O(data) for mimicked S3 rename, O(files) for GCS, O(1) with some occasional pauses on Azure.

There are big challenges here and it goes all the way through the system. There's no point adding a feature in a store if the APIs used to access it don't pick it up; there's no point changing an API if the applications people use don't adopt it.

Which is why input from the people who spent time building object stores and hooking their application is so important. That includes:

• Code
• Bug reports
• Insight from their experiences
• Information about the problems they've had
• Problems their users are having

And that's also why the decision of of the EMR team to isolate themselves from the OSS development holds us back.

We end up duplicating effort, like S3Guard, which is the ASF equivalent of EMR consistent view. The fact that EMR shipped with a feature long before us, could be viewed as their benefit of having a proprietary S3 connector. But S3Guard, like the EMR team, models its work on Netflix S3mper. That's code which one of the EMR team's largest customers wrote, code which Netflix had to retrofit onto the EMR closed-source filesystem using AspectJ.

And that time-to-market edge? Well, is not so obvious any more

1. The EMR S3-optimised committer shipped in November 2018.
2. Its open source predecessor, the S3A committers, shipped in Hadoop 3.1, March 2018. That's over 7-8 months ahead of the EMR implementation.
3. And it shipped in HDP-3.0 in June 2018. That's 5 months ahead.

I'm really happy with that committer, first bit of CS-hard code I've done for a long time, got me into the depths of how committers really work, got an unpublished paper  "A zero Rename Committer" from it. And, in writing the TLA+ spec of the object store I used on the way to convincing myself things worked, I was corrected by Lamport himself.

Much of that commit code was written by myself, but it depended utterly on some insights from Thomas Demoor of WDC, It also contains a large donation of code from Netflix, their S3 committer. They'd bypassed emrfs completely and were using the S3A client direct: we took this, hooked it up to what we'd already started to write, incorporated their mockito tests —and now their code is the specific committer I recommend. Because Netflix were using it and it worked.

A heavy user of AWS S3 wanting to fix their problem, sharing the code, having it pulled into the core code so that anyone using the OSS releases gets the benefit of their code *and their testing*.

We were able to pick that code up because Netflix had gone around emrfs and were writing things themselves. That is: they had to bypass the EMR team. And once they'd one that, we could take it, integrate it and ship it eight months before that EMR team did. With proofs of correctness(-ish).

Which is why I don't believe that isolationism is good for whoever opts out of the community. Nor, by the evidence, is is good for their customers.

I don't even think it helps the EMR team's colleagues with their own support calls. Because really, if you aren't active in the project, those colleagues end up depending on the goodwill of others.

(photo: building in central Havana. There'll be a starbucks there eventually)

How to play with the new S3A committers

Following up from yesterday's post on the S3A committers, here's what you need for picking up the committers.

1. Apache Hadoop trunk; builds to 3.1.0-SNAPSHOT:  
2. The documentation on use.
3. An AWS keypair, try not to commit them to git. Tip for the Uber team: git-secrets is something you can add as a checkin hook. Do as I do: keep them elsewhere.
4. If you want to use the magic committer; turn S3Guard on. Initially I'd use the staging committer, specificially the "directory" on.
5. switch s3a:// to use that committer: fs.s3a.committer.name =  partitioned
6. Run your MR queries
7. look in _SUCCESS for committer info. 0-bytes long: classic FileOutputCommitter. Bit of JSON naming committer, files committed and some metrics (SuccessData) and you are using an S3 committer.

If you do that: I'd like to see the numbers comparing FileOutputCommitter (which must have S3Guard) and the new committers. For benchmark consistency, leave S3Guard on.

If you can't get things to work because the docs are wrong: file a JIRA with a patch. If the code is wrong: submit a patch with the fix & tests.

Spark?

1. Spark Master has a couple of patches to deal with integration issues (FNFE on magic output paths, Parquet being over-fussy about committers, I think the committer binding has enough workarounds for these to work with Spark 2.2 though.
2. Checkout my cloud-integration for Apache Spark repo, and its production-time redistributable, spark-cloud-integration.
3. Read its docs and use
4. If you want to use Parquet over other formats, use this committer.  
5. Again,. check _SUCCESS to see what's going on.
6. There's a test module with various (scaleable) tests as well as a copy and paste of some of the Spark SQL test.
7. Spark can work with the Partitioned committer. This is a staging committer which only worries about file conflicts in the final partitions. This lets you do in-situ updates of existing datasets, adding new partitions or overwriting existing ones, while leaving the rest alone. Hence: no need to move the output of a job into the reference datasets.
8. Problems. File an issue. I've just seen Ewan has a couple of PRs I'd better look at, actually.

Committer-wise, that spark-cloud-integration module is ultimately transient. I think we can identify those remaining issues with committer setup in spark core, after which a hadoop 3.0+ specific module should be able to work out the box with the new committers.

There's still other things there, like

• Cloud store optimised file input stream source. 
• ParallizedWithLocalityRDD: and RDD which lets you provide custom functions to declare locality on a row-by-row basis. Used in my demo of implementing DistCp in Spark. Every row is a filename, which gets pushed out to a worker close to the data, it does the upload. This is very much a subset of distCP, but it shows this: you can have with with RDDs and cloud storage.
• + all the tests

I think maybe Apache Bahir would be the ultimate home for this. For now, a bit too unstable.

(photo: spices on sale in a Mombasa market)

subatomic

I've just committed HADOOP-13786 Add S3A committer for zero-rename commits to S3 endpoints.. Contributed by Steve Loughran and Ryan Blue.

This is a serious and complex piece of work; I need to thank:

1. Thomas Demoor and Ewan Higgs from WDC for their advice and testing. They understand the intricacies of the S3 protocol to the millimetre.
2. Ryan Blue for his Staging-based S3 committer. The core algorithms and code will be in hadoop-aws come Hadoop 3.1.
3. Colleagues for their support, including the illustrious Sanjay Radia, and Ram Venkatesh for letting me put so much time into this.
4. Reviewers, especially Ryan Blue, Ewan Higgs, Mingliang Liu and extra especially Aaron Fabbri @ cloudera. It's a big piece of code to learn. First time a patch of mine has ever crossed the 1MB source barrier

I now understand a lot about commit protocols in Hadoop and Spark, including the history of interesting failures encountered, events which are reflected in the change logs of the relevant classes. Things you never knew about the Hadoop MapReduce commit protocol

1. The two different algorithms, v1 and v2 have very different semantics about the atomicity of task and job commits, including when output becomes visible in the destination directory.
2. Neither algorithm is atomic in both task and job commit.
3. V1 is atomic in task commits, but O(files) in its non-atomic job commit. It can recover from any job failure without having rerun all succeeded tasks, but not from a failure in job commit.
4. V2's job commit is a repeatable atomic O(1) operation, because it is a no-op. Task commits do the move/merge, which are O(file), make the output immediately visible, and as a consequence, mean that failure of a job means the output directory is in an unknown state.
5. Both algorithms depend on the filesystem having consistent listings and Create/Update/Delete operations
6. The routine to merge the output of a task to the destination is a real-world example of a co-recursive algorithm. These are so rare most developers don't even know the term for them -or have forgotten it.
7. At-most-once execution is guaranteed by having the tasks and AM failing when they recognise that they are in trouble.
8. The App Master refuses to commit a job if it hasn't had a heartbeat with the YARN Resource Manager within a specific time period. This stops it committing work if the network is partitioned and the AM/RM protocol fails...YARN may have considered the job dead and restarted it.
9. tasks commit iff they get permission from the AM; thus they will not attempt to commit if the network partitions.
10. if a task given permission to commit does not report a successful commit to the AM; the V1 algorithm can rerun the task; v2 must conclude its in an unknown state and abort the job.
11. Spark can commit using the Hadoop FileOutputCommitter; its Parquet support has some "special" code which refuses to work if the committer is not a subclass of ParquetOutputCommitter
. That is: it's special code makes it the hardest thing to bind to this: ORC, CSV, Avro: they all work out the box.
12. It adds the ability for tasks to provide extra data to its job driver for use in job commit; this allows committers to explicitly pass commit information directly to the driver, rather than indirectly via the (consistent) filesystem.
13. Everyone's code assumes that abort() completes in a bounded time, and does not ever throw that IOException its signature promises it can.
14. There's lots of cruft in the MRv2 codebase to keep the MRv1 code alive, which would be really good to delete

This means I get to argue the semantics of commit algorithms with people, as I know what the runtimes "really do", rather than believed by everyone who has neither implemented part of it or stepped throught the code in a debugger.

If we had some TLA+ specifications of filesystems and object stores, we could perhaps write the algorithms as PlusCal examples, but that needs someone with the skills and the time. I'd have to find the time to learn TLA+ properly as well as specify everything, so it won't be me.

Returning to the committers, what do they do which is so special?

They upload task output to the final destination paths in the tasks, but don't make the uploads visible until the job is committed.

No renames, no copies, no job-commit-time merges, and no data visible until job commit. Tasks which fail/fail to commit do not have any adverse side effects on the destination directories.

First, read S3A Committers: Architecture and Implementation.

Then, if that seems interesting look at the source.

A key feature is that we've snuck in to FileOutputFormat a mechanism to allow you to provide different committers for different filesystem schemas.

Normal file output formats (i.e. not Parquet) will automatically get the committer for the target filesystems, which, for S3A, can be changed from the default FileOutputCommitter to an S3A-specific one. And any other object store which also offers delayed materialization of uploaded data can implement their own and run it alongside the S3A ones, which will be something to keep the Azure, GCS and openstack teams busy, perhaps.

For now though: users of Hadoop can use Amazon S3 (or compatible services) as the direct destination of Hadoop and Spark workloads without any overheads of copying the data, and the ability to support failure recovery and speculative execution. I'm happy with that as a good first step.

(photo: street vendors at the Kenya/Tanzania Border)

Stocator: A High Performance Object Store Connector for Spark

IBM have published a lovely paper on their Stocator 0-rename committer for Spark

Stocator is:

1. An extended Swift client
2. magic in their FS to redirect mkdir and file PUT/HEAD/GET calls under the normal MRv1 __temporary paths to new paths in the dest dir
3. generating dest/part-0000 filenames using the attempt & task attempt ID to guarantee uniqueness and to ease cleanup: restarted jobs can delete the old attempts
4. Commit performance comes from eliminating the COPY, which is O(data),
5. And from tuning back the number of HTTP requests (probes for directories, mkdir 0 byte entries, deleting them)
6. Failure recovery comes from explicit names of output files. (note, avoiding any saving of shuffle files, which this wouldn't work with...spark can do that in memory)
7. They add summary data in the _SUCCESS file to list the files written & so work out what happened (though they don't actually use this data, instead relying on their swift service offering list consistency). (I've been doing something similar, primarily for testing & collection of statistics).

Page 10 has their benchmarks, all; of which are against an IBM storage system, not real amazon S3 with its different latencies and performance.

Table 5: Average run time

	Read-Only 50GB	Read-Only 500GB	Teragen	Copy	Wordcount	Terasort	TPC-DS
Hadoop-Swift Base	37.80±0.48	393.10±0.92	624.60±4.00	622.10±13.52	244.10±17.72	681.90±6.10	101.50±1.50
S3a Base	33.30±0.42	254.80±4.00	699.50±8.40	705.10±8.50	193.50±1.80	746.00±7.20	104.50±2.20
Stocator	34.60±0.56	254.10±5.12	38.80±1.40	68.20±0.80	106.60±1.40	84.20±2.04	111.40±1.68
Hadoop-Swift Cv2	37.10±0.54	395.00±0.80	171.30±6.36	175.20±6.40	166.90±2.06	222.70±7.30	102.30±1.16
S3a Cv2	35.30±0.70	255.10±5.52	169.70±4.64	185.40±7.00	111.90±2.08	221.90±6.66	104.00±2.20
S3a Cv2 + FU	35.20±0.48	254.20±5.04	56.80±1.04	86.50±1.00	112.00±2.40	105.20±3.28	103.10±2.14

The S3a is the 2.7.x version, which has the stabilisation enough to be usable with Thomas Demoor's fast output stream (HADOOP-11183). That stream buffers in RAM & initiates the multipart upload once the block size threshold is reached. Provided you can upload data faster than you run out of RAM, it avoids the log waits at the end of close() calls, so has significant speedup. (The fast output stream has evolved into the S3ABlockOutput Stream (HADOOP-13560) which can buffer off heap and to HDD, and which will become the sole output stream once the great cruft cull of HADOOP-14738 goes in)

That means in the doc, "FU" == fast upload, == incremental upload & RAM storage. The default for S3A will become HDD storage, as unless you have a very fast pipe to a compatible S3 store, it's easy to overload the memory

Cv2 means MRv2 committer, the one which does  single rename operation on task commit (here the COPY), rather than one in task commit to promote that attempt, and then another in job commit to finalise the entire job. So only: one copy of every byte PUT, rather than 2, and the COPY calls can run in parallel, often off the critical path

 Table 6: Workload speedups when using Stocator

	Read-Only 50GB	Read-Only 500GB	Teragen	Copy	Wordcount	Terasort	TPC-DS
Hadoop-Swift Base	x1.09	x1.55	x16.09	x9.12	x2.29	x8.10	x0.91
S3a Base	x0.96	x1.00	x18.03	x10.33	x1.82	x8.86	x0.94
Stocator	x1	x1	x1	x1	x1	x1	x1
Hadoop-Swift Cv2	x1.07	x1.55	x4.41	x2.57	x1.57	x2.64	x0.92
S3a Cv2	x1.02	x1.00	x4.37	x2.72	x1.05	x2.64	x0.93
S3a Cv2 + FU	x1.02	x1.00	x1.46	x1.27	x1.05	x1.25	x0.93

Their TCP-DS benchmarks show that stocator & swift is slower than TCP-DS Hadoop 2.7 S3a + Fast upload & MRv2 commit. Which means that (a) the Hadoop swift connector is pretty underperforming and (b) with fadvise=random and columnar data (ORC, Parquet) that speedup alone will give better numbers than swift & stocator. (Also shows how much the TCP-DS Benchmarks are IO heavy rather than output heavy the way the tera-x benchmarks are).

As the co-author of that original swift connector then, what the IBM paper is saying is "our zero rename commit just about compensates for the functional but utterly underperformant code Steve wrote in 2013 and gives us equivalent numbers to 2016 FS connectors by Steve and others, before they started the serious work on S3A speedup". Oh, and we used some of Steve's code to test it, removing the ASF headers.

Note that as the IBM endpoint is neither the classic python Openstack swift or Amazon's real S3, it won't exhibit the real issues those two have. Swift has the worst update inconsistency I've ever seen (i.e repeatable whenever I overwrote a large file with a smaller one), and aggressive throttling even of the DELETE calls in test teardown. AWS S3 has its own issues, not just in list inconsistency, but serious latency of HEAD/GET requests, as they always go through the S3 load balancers. That is, I would hope that IBM's storage offers significantly better numbers than you get over long-haul S3 connections. Although it'd be hard (impossible) to do a consistent test there, I 'd fear in-EC2 performance numbers to be actually worse than that measures.

I might post something faulting the paper, but maybe I'll should to do a benchmark of my new committer first. For now though, my critique of both the swift:// and s3a:// clients is as follows

Unless the storage services guarantees consistency of listing along with other operations, you can't use any of the MR commit algorithms to reliably commit work. So performance is moot. Here IBM do have a consistent store, so you can start to look at performance rather than just functionality. And as they note, committers which work with object store semantics are the way to do this: for operations like this you need the atomic operations of the store, not mocked operations in the client.

People who complain about the performance of using swift or s3a as a destination are blisfully unaware of the key issue: the risk of data loss due inconsistencies. Stocator solves both issues at once.

Anyway, means we should be planning a paper or two on our work too, maybe even start by doing something about random IO and object storage, as in "what can you do for and in columnar storage formats to make them work better in a world where a seek()+ read is potentially a new HTTP request."

(picture: parakeet behind Picton Street)

The Great S3 Outage of February 2017

On tuesday the world split into different groups

1. Those who knew that S3 was down, and the internet itself was in crisis.
2. Those who knew that some of the web sites and phone apps they used weren't working right, but didn't know why.
3. Those who didn't notice and wouldn't have cared.

I was obviously in group 1, the engineers, who whisper to each other, "where were you when S3 went down".



I was running the latest hadoop--aws s3a tests, and noticed as some of my tests were failing. Not the ones to s3 Ireland, but those against the landsat bucket we use in lots of our hadoop test as it is a source of a 20 MB CSV file where nobody has to pay download fees, or spend time creating a 20 MB CSV file. Apparently there are lots of landsat images too, but our hadoop tests stop at: seeking in the file. I've a spark test which does the whole CSV parse thing., as well as one I use in demos as an example not just of dataframes against cloud data, but of how data can be dirty, such as with a cloud cover of less than 0%.

Partial test failures: never good.

It was only when I noticed that other things were offline that I cheered up: unless somehow my delayed-commit multipart put requests had killed S3: I wasn't to blame. And with everything offline I could finish work at 18:30 and stick some lasagne in the oven. (I'm fending for myself & keeping a teenager fed this week).

What was impressive was seeing how deep it went into things. Strava app? toast. Various build tools and things? Offline.

Which means that S3 wasn't just a SPOF for my own code, but a lot of transitive dependencies, meaning that things just weren't working -all the way up the chain.



S3 is clearly so ubiquitous a store that the failure of US-East enough to have major failures, everywhere.

Which makes designing to be resilient to an S3 outage so hard: you not only have to make your own system somehow resilient to failure, you have to know how your dependencies cope with such problems. For which step one is: identify those dependencies.

Fortunately, we all got to find out on Tuesday.

Trying to mitigate against a full S3A outage is probably pretty hard. At the very least,

1. replicated front end content across different S3 installations would allow you to present some kind of UI.
2. if you are collecting data for processing, then a contingency plan for the sink being offline: alternate destinations, local buffering, discarding (nifi can be given rules here).
3. We need our own status pages which can be updated even if the entire infra we depend on is missing. That is: host somewhere else, have multiple people with login rights, so an individual isn't the SPOF. Maybe even a facebook page too, as a final backup
4. We can't trust the AWS status page no more.

Is it worth putting in lots of effort to eliminating an S3 outage as a SPOF? Well, the failure rate is such that it's a lot of effort for a very rare occurence. If you are user facing, some app like strava, maybe it's easiest to say "no". If you are providing a service for others though, availability, or at least the ability to degrade QoS is something to look at.

Anyway, we can now celebrate the fact that the entire internet now runs in four places: AWS, Google, Facebook and Azure. And we know what happens when one of them goes offline.

How long does FileSystem.exists() take against S3?

One thing I've been working on with my colleagues is improving performance of Hadoop, Hive and Spark against S3, one exists() or getFileStatus() call at a time.

Why? This is a log of a test run showing how long it takes to query S3 over a long haul link. This is midway through the test, so the HTTPS connection pool is up, DNS has already resolved the hostnames. So these should be warm links to S3 US-east. Yet it takes over a second just for one probe.

2016-12-01 15:47:10,359 - op_exists += 1  ->  6
2016-12-01 15:47:10,360 - op_get_file_status += 1  ->  20
2016-12-01 15:47:10,360 (S3AFileSystem.java:getFileStatus) -
  Getting path status for s3a://hwdev-stevel/numbers_rdd_tests
2016-12-01 15:47:10,360 - object_metadata_requests += 1 -> 39
2016-12-01 15:47:11,068 - object_metadata_requests += 1 -> 40
2016-12-01 15:47:11,241 - object_list_requests += 1 -> 21
2016-12-01 15:47:11,513 (S3AFileSystem.java:getFileStatus) -
  Found path as directory (with /)

The way we check for a path p in Hadoop's S3 Client(s) is

HEAD p
HEAD p/
LIST prefix=p, suffix=/, count=1

A simple file: one HEAD. A directory marker, two, a path with no marker but 1+ child: three. In this run, it's an empty directory, so two of the probes are executed:

HEAD p => 708ms
HEAD p/ => 445ms
LIST prefix=p, suffix=/, count=1 => skipped

That's 1153ms from invocation of the exists() call to it returning true —long enough for you to see the log pause during the test run. Think about that: determining which operations to speed up not through some fancy profiler, but watching when the log stutters. That's how dramatic the long-haul cost of object store operations are. It's also why a core piece of the S3Guard work is to offload that metadata storage to DynamoDB. I'm not doing that code, but I am doing the committer to go with. To be ruthless, I'm not sure we can reliably do that O(1) rename, massively parallel rename being the only way to move blobs around, and the committer API as it stands precluding me from implementing a single-file-direct-commit committer. We can do the locking/leasing in dynamo though, along with the speedup.

What it should really highlight is that an assumption in a lot of code "getFileStatus() is too quick to measure" doesn't hold once you move into object stores, especially remote ones, and that any form of recursive treewalk is potentially pathologically bad.
Remember that that next time you edit your code.

Gardening the commons

It's been a warm-but-damp summer in Bristol, and the vegetation in the local woods has been growing well. This means the bramble and hawthorn branches have been making their way out of the undergrowth and into the light —more specifically the light available in the mountain bike trails.

Being as both these plant's branches have spiky bits on them, the fact that they are growing onto the trails hurts, especially if you are trying to get round corners fast. And, if anyone is going round the trail without sunglasses they run a risk of getting hurt in/near the eye.

I do always wear sunglasses, but the limitations on taking the fast line though the trails hurts, and as there a lot of families out right now, I don't want the kids to get too scraped.

So on Saturday morning, much to the amazement of my wife, I picked up the gardening shears. Not to do anything in our garden though —to take to the woods with me.



Findings

1. Those Kevlar backed gloves that the Clevedon police like are OK on the outside for working with spiky vegetation, but the fingertips are vulnerable.
2. Gardening gets boring fast.
3. When gardening an MTB trail, look towards the direction oncoming riders will take.
4. A lot of people walking dogs get lost and ask for directions back to the car park.
5. Someone had already gardened about 1/3 of the Yertiz trail. (pronounciation based on: "yeah-it-is"
6. Nobody appreciates your work.

I appreciate the outcome my own work, I can now go round at speed, only picking up scrapes on the forearms on the third of the trail that nobody has trimmed yet. I actually cut back on the inside of the corners there for less damage on the racing line, while cutting the outside and face height bits for the families. Now they can have more fun on the weekends, I can do my fast work weekday lunchtimes.


They can live their lives with fewer wailing children, and I've partially achieved my goal of less blood emitted per iteration. I'll probably finish it off with the final 1/3 at some point, maybe mid-august, as I can't rely on anyone else.

There are no trail pixies but but what we make



Which brings me to OSS projects, especially patches and bug reports in Hadoop.

I really hate it when my patches get completely ignored, big or small.

Take YARN-679 for example. Wrap up piece of YARN-117, code has celebrated its third birthday. Number of reviewers. Zero. I know its a big patch, but it's designed to produce a generic entry point for YARN services with reflection based loading of config files (you can ask for HiveConfig and HBaseConfig too, see), interrupt handling which even remembers if its been called once, so on the second control-C bypasses the shutdown hooks (assumption: they've blocked on some IPC-retry chatter with a now-absent NN), and bail out fast. Everything designed to support exit codes, subclass points for testability. This should be the entry point for every single YARN service, and it hasn't had a single comment by anyone. How does that make me feel. Utterly Ignored —even by colleagues. I do, at least, understand what else they are working on...it's not like there is a room full of patch reviewers saying "there are no patches to review —let's just go home early". All the people with the ability to review the patches have so many commitments of their own, that the time to review they can allocate is called "weekends".

And as a result, I have a list of patches awaiting review and commit, patches which are not only big diffs, they are trivial ones which fix things like NPEs in reporting errors returned over RPC. That's a 3KB patch, reaching the age where, at least with my own child, we were looking at nursery schools. Nothing strategic, something useful when things go wrong. Ignored.

That's what really frustrates me: small problems, small patches, nobody looks at it.

And I'm as guilty as the rest. We provide feedback on some patch-in-progress, then get distracted and never do the final housekeeping. I feel bad here, precisely because I understand the frustration.

Alongside "old, neglected patches", there are "old, neglected bugs". Take, for an example HADOOP-3733 "s3:" URLs break when Secret Key contains a slash, even if encoded. Stuart Sierra gave a good view of the history from his perspective.

The bug was filed in 2008

1. it was utterly ignored
2. Lots of people said they had the same problem
3. Various hadoop developers said "Cannot reproduce"
4. It was eventually fixed on 2016-06-16 with a patch by one stevel@apache.
5. On 2016-06-16 cnauroth@apache filed HADOOP-13287 saying "TestS3ACredentials#testInstantiateFromURL fails if AWS secret key contains '+'.".

Conclusions

• The Hadoop developers neglect things
• if we'd fixed things earlier, similar problems won't arise.

I mostly concur. Especially in the S3 support, where historically the only FTEs working on it were Amazon, and they have their own codebase. In ASF Hadoop, Tom White started the code, and it's slowly evolved, but it's generally been left to various end users to work on.

Patch submission is also complicated by the fact that for security reasons, Jenkins doesn't test the stuff. We've had enough problems of people under-testing their patches here that there is a strictly enforced policy of "tell us which infrastructure you tested against". The calling out of "name the endpoint" turns out to be far better at triggering honest responses than "say that you tested this". And yes, we are just as strict with our colleagues. A full test run of the hadoop-aws module takes 10-15 minutes, much better than the 30 minutes it used to take, but still means that any review of a patch is time consuming.

I would normally estimate the time to review an S3 patch to take 1-2 hours. And, until a few of us sat down to work on S3A functionality and performance against Hive and Spark, those 1-2 hours were going weekend time only. Which is why I didn't do enough reviewing.

Returning to the S3 "/", problem

1. This whole thing was related to AWS-generated secrets. Those of us whose AWS secrets didn't have a "/" in this couldn't replicate the problem. Thus it was a configuration-space issue rather than something visible to all.
2. There was a straightforward workaround, "generate new credentials", so it wasn't a blocker.
3. That related issue, HADOOP-13287, is actually highlighting a regression caused by the fix for HADOOP-3733. In the process for allowing URLs to contain "/" symbols, we managed to break the ability to use "+" in them. 
4. The regression was caught because the HADOOP-3733 patch included some tests which played with the tester's real credentials. Fun problem: writing tests to do insecure things which don't leak secrets in logs and assert statements.
5. HADOOP-13287 is not an example of "there are nearby problems" so much as "every bug fix moves the bug", something noted in Brook's "the mythical man month" in his coverage of IBM OS patches.
6. And again, this is a c-space problem, it was caught because Chris had + in his secret.

Finally, and this is the reason why it didn't surface with many of us, even though we had "/" in the secret is because the problem only arises if you put your AWS secrets in the URL itself, as s3a://AWSID:secret-aws-key@bucket

That is: if your filesystem URI contains secrets, which, if leaked —threaten the privacy and integrity of your data and is at risk of running up large bills, then, if the secret has a "/", the URL doesn't authenticate.

This is not actually an action I would recommend. Why? Because throughout the Hadoop codebase we assume that filesystem URIs do not contain secrets. They get logged, they get included in error messages, they make their way into stack traces that can go into bug reports. AWS credentials are too important to be sticking in URLs.

Once I realised people were doing this, I did put aside a morning to fix things. Not so much fixing the encoding of  "/" in the secrets (and accidentally breaking the encoding of "+" in the process), but:

1. Pulling out the auth logic for s3, s3n and s3a into a new class, S3xLoginHelper.
2. Having that code strip out user:pass from the FS URL before the S3 filesystems pass it up to their superclass.
3. Doing test runs and seeing if that is sufficient to keep those secret out the logs (it isn't).
4. Having S3xLoginHelper print a warning whenever somebody tries to use secrets in URLs.
5. Edit the S3 documentation file to tell people not to do this —and warning the feature may be removed.
6. Edit the Hadoop S3 wiki page telling people not to do this.
7. Finally: fix the encoding for /, adding tests
8. Later, fix the test for +

That's not just an idle "may be removed" threat. In HADOOP-13252, you can declare which AWS credential providers to support in S3A, be it your own, conf-file, env var, IAM, and others. If you start doing this, your ability to embed secrets in s3a URLs goes away. Assumption: if people know what they are doing, they shouldn't be doing things so insecure.

Anyway, I'm glad my effort fixing the issue is appreciated. I also share everyone's frustration with neglected patches, as it wastes my effort and leaves the bugs unfixed, features ignored.

We need another bug bash. And I need to give that final third of the Yertiz trail a trim.

Testing against S3 and object stores —keep your keys out of SCM

Nobody wants to commit their AWS credentials to an SCM repository, not when there are robots looking for your keys so they can spin up bitcoin mining VMs.


That's why git-secrets is something you should preinstall on all repos where you go near repositories. Unfortunately, it's pattern matching picks up a lot of invalid patterns in the Hadoop codebase, and a few in spark. In the absence of a decent pattern to only scan text files, after installing I edit out the regexps from .git/config and rely on it scanning purely for the strings in ~/.aws/credentials.

That keeps the keys out of SCM, if everyone is set up that way.

Which leaves the next problem: if you save your hadoop configuration files to SCM, how do you get AWS keys into system configurations?


The answer: XInclude


This is one of my configuration files, specifically hadoop-tools/hadoop-aws/src/test/resources/auth-keys.xml

This is a special file, tagged as .gitignore to keep it out of the repos. Yet still the keys are in that source tree, still at risk of sneaking out.

To deal with this, stick an absolute XInclude reference into the file, pointing to the configuration file where the keys really live.

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

This tells the Hadoop config loader to grab the keys from a file in ~/.aws; one which lives carefully out of the SCM-managed space.

Provided the contents of that directory are kept private, my keys will not get checked in.

They can, however leak, in various ways, including

• In the /config URL of a service
• Code which accidentally logs it.
• If it gets saved to files/datasets used in bug reports.
• Malicious code running in your system which grabs the keys and exports/logs them. This is why no OSS jenkins servers are set up with the keys needed to test against object store.

Clearly, more should be done to lock it down.

There are some patches under the S3A phase 2 JIRA to support credentials better, [HADOOP-12723, HADOOP-12537]. This is somewhere where anyone who can test these patches is invaluable. I'm currently doing the review-then-commit in S3, but I'm only doing it at weekends, when I have some spare time, and even then, as a full test run takes 2+ hours, not reviewing very much.

Anyone can review Hadoop patches, confirm whether they worked or not, show how they didn't. It's the way to verify that forthcoming code from other people works for you, and of contributing time and effort into the community. Other people have done the coding —time to help with the testing.

Especially now I've just documented how to keep the keys safe when you set up S3 for a test run. Once the auth-keys.xml file is in the src/test/resources directory, Maven tests s3, depending on the specific properties, s3a, s3n and s3a are all tested.

Update: 2016-06-20: I've discovered that if you use the id:secret in a URL for an S3x filesystem, e.g s3a://AWS01:sec/ret@mybucket then the secrets get logged everywhere, because the Hadoop code assumes there aren't secrets in the Path or Filesystem URIs. We've cranked back a bit on the leakage (by stripping it from the FS URIs), but it still gets out everywhere. Fix: don't do that. Hadoop 2.8 will explicitly tell you off whenever you do so.

Update: 2018-02-07: Hadoop 2.9+ uses a more efficient Stax XML parser over a DOM based one. That broke this secret-sequestering somewhat dramatically. For Hadoop 2.9+, do not use the file: prefix

  <include xmlns="http://www.w3.org/2001/XInclude"
    href="///users/stevel/.aws/auth-keys.xml" >
  </include>

FWIW, it's possible to have base auth-key.xml with XIncludes which supports both parsers by having both references in the file, and add a <fallback/> element in both so an unresolvable ref is a no-op. Worth knowing if you need to test across all branches.

[photo: Mina Road Park, St Werburghs]

Defeated by iPad synchronization options

Last January I got an iPad mini as a travel accessory to the laptop: music, eBooks, PDF formatted papers, online and offline maps, etc.



It's also intended to be holder of travel paperwork: the schedule, logistics notes, eTickets, hotel details. All mostly PDF, though my KLM check in has just emailed a GIF QF barcode which apparently will get me through security (outbound I'm testing w/ a backup paper one, return: commit to GIF).

A major use case of mine then is: get PDFs off my laptop and into the iPad so that I can bring them up and view them.

Which is where it all seems to go horribly wrong.

I can see four different synchronization options.

iTunes
Copy into the books section of iTunes and let them trickle over via USB or wifi.
this works, provided the devices can see each other in the same wifi subnet.

It is a bit clunky as I have to drag and drop content from my folder of travel bits (e.g. 2013-03-AMS) and store them in a flat pool of documents, where they end up mixed next to things like Grinstead and Cell's Introduction to Probability, papers on things like Chubby, and copies of Singletrack Magazine. This isn't ideal for navigating at the airport security gate.

But again: it works, and I know how to verify that the stuff has trickled over -you look at the sync status page.

Workflow

1. Clean up the last trips' documents.
2. Copy in the new files
3. Force a sync to make sure it is over.
4. Updates: steps 2 and 3.

Apple iCloud

This is meant to be the future. Instead of saving to the filesystem, you save it to "the cloud" where it will magically make its way over to your other devices.

Except I put PDFs in there and there doesn't seem to be any obvious way to actually see that they have made it over, let alone open it.

This is not a Cloud, it is /dev/null with unrealistic promises. I could say that if I copy a file to /dev/null then all my other devices will get the same view of the copied documents -but if they aren't there, it's not a very good view.

The workflow for getting documents over via iCloud is therefore

1. Save the files into iCloud.
2. Pick one of the other synchronization options to get your content over.

Dropbox

The folder metaphor, I can drag and drop anything in on my desktop to it, it trickles over across all my desktops, OS/X and linux.

What it doesn't do is automatically trickle the files over to the iPad. It copies the directory metadata over, but I seem to have to tap every file -by hand- for it to decide that it's going to download every artifact in the filesystem.

For anyone with a default 2 GB Dropbox account, you could copy everything over while on wifi and not use up any device space, even for customers like me who went for the low-end 16GB model because they felt the cost/GB of extra SSD in an iPad was utterly excessive.

The workflow to sync is therefore

1. Save all the content into a dropbox managed folder 
2. go to the tablet and find that folder
3. go to every file in it, and manually hit the download button, wait for it to D/L. Repeat for all files in a process that is O(files)*O(filesize).
4. The update process is steps 1-3, repeated.

Box
I also have a box account, and an iPad app for that. This has some flags about auto-syncing on wifi only, which I'm happy with, not having a device with a modem in (tethering & wifi usually suffice).

It also has -and this got me excited- the ability to mark folders as "favorite", where it is claimed that content will auto-sync to the pad. I was hopeful here, marked my folder travel as fave and then put stuff into trip-specific subdirs underneath, for this week's trip, next month's US trip, and others.

I go over to the 'pad, expecting the files to be there.

Only they aren't, because the favoured bit is not recursive.

Once you know that, Box sync becomes manageable

1. Create a folder for each planned trip.
2. Copy travel docs in there.
3. Go to your table, and mark that folder as favourite, even if the parent dir is already marked as such.
4. Update the travel folder on the laptop -things will now trickle over.

Remember step #3 and it does work.


Email
Just mail them to yourself the day before you travel, download the mail and make sure the files are there.

This appears to work, though there are probably limits on how big the files are that are auto-D/L'd, and the files go away at the history rate specified in the mail app, which is no good for a long trip.

Workflow

1. Make sure the mail app is set to cache data for >= the length of your trip.
2. Email the PDFs to yourself.
3. Verify in the mail app that they have all arrived.

The nice feature about this is that it works from everywhere, whether or not box is installed. You can also get other people to contribute to the document pool by having them email you direct.

There you have it: ways I've tried to sync documents.



If iCloud actually seemed to do what is promised "share your content across devices via Apple's cloud" then it might work, even if it's metaphor "not a filesystem, but a place where every artefact is permanently bonded to whichever application put it there, even if there is >1 text or PDF viewer on a device", is so dumbed down it represents a step back to Mac 1.0.

Unfortunately the behaviour I see "the same consistency and durability model as copying files to /dev/null", means that there is no way I would trust it with anything. I actually hope there may be something obvious I'm missing here, as I can't understand how something so dire would spring into existence, and don't believe the business plan "make money from premium users who want to store more stuff" stands a chance against tools that actually work.

Instead I've settled on Box, making sure that things go over (opening a non-random sample of them -I should toss a coin over each file for better randomness.

Oh, and print out my boarding card and the map from the train station to the hotel.

[Photos: some of the CCTVs I saw on a single walk down Gloucester Road. I'm not quite sure what problems this high street had that needed near-ubiquitous CCTV coverage, but there's enough cameras to have fixed it. I like the one pointed straight at the ATM the best]

AWS: why the bias towards US-east?

As MastodonC will point out, Amazon's US-East sites are the most polluting, not just because they have a high CO2 footprint, but because the coal they (and the other east coast) industries burn is polluting in other ways, such as sulphur. It's not as bad as, say, a steelworks (having had relatives living near  Ravenscraig Steelworks I can vouch for this), but as datacentres can be placed near other electricity sources, it's needless.

I intermittently use US-West-2, up in Oregon, where the melting snow creates electricity.



Unfortunately there's an implicit bias in the AWS APIs towards US-East. Where's the default site for S3 Buckets? US-East. Where's the default site for EC2 instances? US-East. What is the default location for EMR jobs? The same -to the extent that the command line clients treats requesting a different site as "uncommon":

Uncommon Options
 --debug               Print stack traces when exceptions occur
 --endpoint ENDPOINT   EMR web service host to connect to
 --region REGION       The region to use for the endpoint
 --apps-path APPS_PATH Specify s3:// path to the base of the emr public bucket to use. e.g s3://us-east-1.elasticmapreduce

Because of all the implicit "us-east" bias, it becomes self reinforcing. Once you've got a bucket on S3 east, that's where you want to run your webapps otherwise you get billed for the remote bandwidth. Once you've got the webapps, that's where your logs go, hence even more reason to run your MR jobs on the same site: it's where your data lives.

Because it's the default location for stuff, it's also the default location for people serving up data on the site: RPM and Maven repositories, public datasets. This pushes you towards that location so as to avoid the costs of downloading that data from other sites, as well as the speed gain.

Why the bias? Either it's where the the majority of servers lie, or through a combination of cost of electricity, site PUE and bandwidth, it's got the lowest operating costs -hence the most profit per CPU-hour, MB stored or MB downloaded.

That's a shame, because amazon themselves have better options. They're being crucified by Parliament over their tax avoidance strategies -it'd be tactically wise to have something positive to talk about.

[Photo: Crater Lake & Mt Thielsen. Smoke is a forest fire blowing up from CA]

Welcome to Chaos

Netflix have published their "Chaos Monkey" code on Github; ASL Licensed. I have already filed my first issue, having looked through the code -an issue that is already marked as fixed.

Netflix bring to the world the original Chaos Monkey -tested against production services.

Those of us playing with failures, reliability and availability in the Hadoop world also need something that can generate failures, though for testing the needs are slightly different:

1. Failures triggered somewhat repeatedly.
2. Be more aggressive.
3. Support more back ends than Amazon -desktop, physical, private IaaS infrastructures.

#1 and #2 are config tuning -faster killing, seeded execution.

#3? Needs more back ends. The nice thing here is that there's very little you need to implement when all you are doing is talking to an Infrastructure Service to kill machines; the CloudClient interface has one method:

  void terminateInstance(String instanceId);

That needs to be aided with something to produce a list of instances, and of course there's the per-infrastructure configuration of URLs and authentication.


My colleague Enis has been doing something for this for HBase testing.; independently I've done something in groovy for my availability work, draft package, org.apache.chaos. I've done three back ends :

1. SSH in to a machine and kill a process by pid file.
2. Pop up a dialog telling the user to kill a machine (not so daft, good for semi-automated testing).
3. Issue virtualbox commands to kill a VM.

All of these are fairly straightforward to migrate to the Chaos Monkey; they are all driven by config files enumerating the list of target machines, plus some back-end specific options (e.g. pid file locations, list of vbox UUIDs).

Then there's the other possibilities: VMWare, fencing devices on the LAN, ssh in and issue "if up/down" commands (though note that some infrastructures, such as vSphere, recognise that explicit option and take things off HA monitoring). All relatively straightforward. 

Which means: we can use the Chaos Monkey as a foundation for testing how distributed systems, especially the Hadoop stack components, react to machine failover -across a broad set of virtual and physical infrastructures. 


That I see the appeal of.


Because everyone needs a Chaos Monkey.

[update 13:24 PST, fixed first name, thank you Vanessa Alvarez!]

Hadoop in Cloud Infrastructures

People say "should you run Hadoop in the cloud?". I say "it depends".

I think there is value in Hadoop-in-cloud, I talked about doing it in 2010 at Berlin Buzzwords 2010; since then I've had more experience with using Hadoop and implementing cloud infrastructures.

1. If your data is stored in a cloud provider's storage infrastructure, doing the analysis locally is the only rational action. It's that "work near the data" philosophy.
2. If you are only doing some computation -say nightly- then you can rent some cluster time. Even if compute performance is worse, you can just rent some more machines to compensate.
3. You may be able to achieve better security through isolation of clusters (depends on your IaaS vendor's abilities).
4. No upfront capex; fund from ongoing revenue.
5. Easier to expand your cluster; no need to buy more racks, find more rack space.
6. You don't need to care about the problems of networking.
7. Less of a problem of heterogenous clusters if you expand later.

Against that

1. Cost of data storage can only increase at a rate proportional to ingress/retention rates.
2. Cost of cluster time increases at a rate proportional to analysis performed. There is no "spare cluster time" for low priority work.
3. Even if CPU time can scale up, IO rate of persistent data may not.
4. Hadoop contains lots of assumptions about running in a static infrastructure; it's scheduling and recovery algorithms assume this.

Some examples of where Hadoop's assumptions diverge from that of cloud infrastructures:

• HDFS assumes failures are independent, and places data accordingly (Google's Availability in Globally Distributed Storage Systems paper shows this doesn't hold in physical infrastructures, my notion of failure topologies expands on that)
• MR blacklists failing machines, rather than releasing them and requesting new ones.
• Worker nodes handle failure of master nodes by spinning on the hostname, not querying (dynamic) configuration data for new hostnames. Some of the HA HDFS may address that, I'm not tracking it enough.
• Topology scripts are static. I've been slowly tweaking topology logic in 0.23+ but haven't put the dynamicness in there yet (HDFS and MR cache (name->rack) mappings on the assumption that the data is coming from slow to exec scripts, not fast & refreshable in-VM data).
• Schedulers assume #of machines are static, don't allocate and release compute nodes based on demand and with knowledge of cost and quantum of CPU rentals. (I'm not sure quantum is the right term there, I mean the fact that VMs may be rented by the hour, 15 minutes, etc, so your scheduler should retain them for 59 minutes after acquiring them.
• Scheduling doesn't bill different users for their cluster use in a way that is easily mapped to cluster time.

A lot of these are tractable, you just have to put in the effort. The Stratosphere team in Berlin are doing lots of excellent work here, including taking a higher level query language and generating an execution plan that is IaaS aware -you can optimise for fast (many machines) or lower cost (use less machines more efficiently).

In comparison, a physical cluster:

• Offers a lower cost/TB of any corporate filestore to date other than people's desktop computers (which have a high TCO and power cost that is generally ignored), so enables you to store lots of stuff you would otherwise discard.
• Let's you choose the hardware optimised for your current and predicted workloads.
• Has free time for the low priority background work as well as the quicker queries that near-real-time UIs like.
• May be directly accessible from desktops in the organisation (depends on security model of cluster).
• Is easily hooked up to Jenkins infrastructure for execution of work as CI jobs.
• Let's you do fancy tricks like striping of different MR versions across the racks for in-rack locality and different sets of task trackers for foreground vs background work, and different JTs (reduces memory use, cost of failure, etc).
• Is way, way easier to hook up to internal databases, log feeds. To do ETL into your corporate oracle servers, you will need to run something behind the firewall to fetch it off the IaaS storage layer, rather than have your reducers push it to the RDBMS itself.

If you are generating data in house, in house clusters make a lot of sense.
This is why I say "it depends" -it depends on where you collect your data and what you plan to do with it.

As for the way Hadoop doesn't currently work so well in such infrastructures, well the code is there for people to fix. It's also a lot easier to test in-cloud behaviour, including resilience to failure, than it is with physical clusters.

[Photo: Descending Mt Ranier, 2000]