Dancing elephants - efficiently working with object stores from Apache Spark and Apache Hive

Editor's Notes

• #7: on prem = 1 azure @ 3 AWS @ 2
• #8: In all the examples, object stores take a role which replaces HDFS. But this is dangerous, because...
• #9: This is one of the simplest deployments in cloud: scheduled/dynamic ETL. Incoming data sources saving to an object store; spark cluster brought up for ETL. Either direct cleanup/filter or multistep operations, but either way: an ETL pipeline. HDFS on the VMs for transient storage, the object store used as the destination for data —now in a more efficient format such as ORC or Parquet
• #10: Notebooks on demand. ; it talks to spark in cloud which then does the work against external and internal data; Your notebook itself can be saved to the object store, for persistence and sharing.
• #11: Example: streaming on Azure + on LHS add streaming
• #12: In all the examples, object stores take a role which replaces HDFS. But this is dangerous, because...
• #13: This pictureis classic HDFS – 3 replicas and metadata First off all there are no real directories in S3 – the directory names are part of the file name – Cloud object store does not treat “/” as a special separator. Here in the rename all you expect is that the data remains and simply some metadata change – that is what happens in HDFS but no so in S3
• #14: There is no rename operation – a copy and delete has to be done The copied data is new (Yellow) The deleted data (Orange)
• #15: You might say – big deal the data in pending is deleted later – but imaging if you did direct output connector and was deleting failed task output Also same problem can occur for new data you added.
• #16: We will come back to the inconsistency problem and the new S3Guard
• #17: Everything uses the Hadoop APIs to talk to both HDFS, Hadoop Compatible Filesystems and object stores; the Hadoop FS API. There's actually two: the one with a clean split between client side and "driver side", and the older one which is a direct connect. Most use the latter and actually, in terms of opportunities for object store integration tweaking, this is actually the one where can innovate with the most easily. HDFS is "real" filesystem; WASB/Azure close enough. What is "real?". Best test: can support HBase.
• #18: This is the history
• #20: Simple goal. Make ASF hadoop at home in cloud infra. It's always been a bit of a mixed bag, and there's a lot with agility we need to address: things fail differently. Step 1: Azure. That's the work with Microsoft on wasb://; you can use Azure as a drop-in replacement for HDFS in Azure Step 2: EMR. More specifically, have the ASF Hadoop codebase get higher numbers than EMR
• #22: Here's a flamegraph of LLAP (single node) with AWS+HDC for a set of TPC-DS queries at 200 GB scale; we should stick this up online only about 2% of time (optimised code) is doing S3 IO. Something at start partitioning data
• #23: without going into the details, here are things you will want for Hadoop 2.8. They are in HDP 2.5, possible in the next CDH release. The first two boost input by reducing the cost of seeking, which is expensive as it breaks then re-opens the HTTPS connection. Readahead means that hundreds of KB can be skipped before that connect (yes, it can take that long to reconnect). The experimental fadvise random feature speeds up backward reads at the expense of pure-forward file reads. It is significantly faster for reading in optimized binary formats like ORC and Parquet The last one is a successor to fast upload in Hadoop 2.7. That buffers on heap and needs careful tuning; its memory needs conflict with RDD caching. The new version defaults to buffering as files on local disk, so won't run out of memory. Offers the potential of significantly more effective use of bandwidth; the resulting partitioned files may also offer higher read perf. (No data there, just hearsay).
• #24: If you look at what we've done, much of it (credit to Rajesh & Chris) is just minimising HTTP requests. Each one can take hundreds of millis, sometimes even seconds due to load balancer issues (tip: reduce DNS TTL on your clients to <30s). A lot of the work internal to S3A was culling those getFileStatus() calls by (a) caching results, (b) not bothering to see if they are needed. Example: cheaper to issue a DELETE listing all parent paths than actually looking to see if they exist, wait for the response, and then delete them. The one at the end, HADOOP-13208, replaces a slow recursive tree walk (many status, many list) with a flat listing of all objects in a tree. This works only for the listStatus(path, true) call —to be benefit you need to use that API call, not do your own treewalk.
• #25: Don't run off saying "hey, 2x speedup". I'm confident we got HDP faster, EMR is still something we'd need to look at more. Data layout is still a major problem here; I think we are still understanding the implications of sharding and throttling. What we do know is that deep/shallow trees are pathological for recursive treewalks, and they end up storing data in the same s3 nodes, so throttling adjacent requests.
• #26: And the result. Yes, currently we are faster in these benchmarks. Does that match to the outside world? If you use ORC & HIve, you will gain from the work we've done. There are still things which are pathologically bad, especially deep directory trees with few files 26x faster!!!
• #29: This invariably ends up reaching us on JIRA, to the extent I've got a document somewhere explaining the problem in detail. It was taken away because it can corrupt your data, without you noticiing. This is generally considered harmful.
• #30: Recap – S3 is eventually cinsistent