Re: Observer Namenode and Committer Algorithm V1

[Adam Binford <ad...@gmail.com>]

Sharing some things I learned looking into the Delta Lake issue:

- This was a read after write inconsistency _all on the driver_.
Specifically it currently uses the FileSystem API for reading table logs
for greater compatibility, but the FileContext API for writes for atomic
renames. This led to the FileSystem reads becoming stale as they didn't
have to update their state ID after the FileContext writes from a different
DFSClient.
- The FileContext API generally seems less suitable for an HA HDFS setup,
as each FileContext object creates a new DFSClient that has to re-find the
active/observer nodes. I know these are cheap operations, but still extra
overhead and not ideal. This is compounded by the fact that the name
"FileContext" is misleading, as it sounds like something you should create
a new instance of per file you want to interact with, and not try to reuse
across a file system. There's been an open issue for 12 years about adding
caching to the FileContext/AbstractFileSystem API, but there seems some
hesitation there due to what happens when you update HDFS while a client is
still active. This doesn't appear to be a huge issue directly in spark,
since the main place FileContext is used is for the structured streaming
commit log for atomic renames, but something to look out for in third party
libraries. I do see a lot of warnings about the HDFSMetadataLog looking for
the active namenode which I haven't looked into much. I'd expect to only
see that once since it seems to properly reuse a single FileContext
instance.

Adam

On Fri, Aug 20, 2021 at 2:22 PM Steve Loughran <st...@cloudera.com.invalid>
wrote:

>
> ooh, this is fun,
>
> v2 isn't safe to use unless every task attempt generates files with
> exactly the same names and it is okay to intermingle the output of two task
> attempts.
>
> This is because task commit can felt partway through (or worse, that
> process pause for a full GC), and a second attempt committed. Spark commit
> algorithm assumes that it's OK to commit a 2nd attempt if the first attempt
> failed, timed out etc. It is for v1, but not v2
>
> Therefore: a (nonbinding) -1 to any proposal to switch to v2. You are only
> changing problems
>
>
> I think the best fix here is to do it in the FileOutputCommitter. Be aware
> that we are all scared of that class and always want to do the minimum
> necessary.
>
> I will certainly add to the manifest committer, whose "call for reviewers
> and testing" is still open, especially all the way through spark
> https://github.com/apache/hadoop/pull/2971
>
> That committer works with HDFS too, I'd be interested in anyone
> benchmarking it on queries with deep/wide directory trees and with
> different tasks all generating output for the same destination directories
> (i.e file rename dominates in job commit, not task rename). I'm not
> optimising it for HDFS -it's trying to deal with cloud storage quirks like
> nonatomic dir rename (GCS), slow list/file rename perf (everywhere), deep
> directory delete timeouts, and other cloud storage specific issues.
>
>
> Further reading on the commit problem in general
> https://github.com/steveloughran/zero-rename-committer/releases/tag/tag_release_2021-05-17
>
> -Steve
>
>
>
> On Tue, 17 Aug 2021 at 17:39, Adam Binford <ad...@gmail.com> wrote:
>
>> Hi,
>>
>> We ran into an interesting issue that I wanted to share as well as get
>> thoughts on if anything should be done about this. We run our own Hadoop
>> cluster and recently deployed an Observer Namenode to take some burden off
>> of our Active Namenode. We mostly use Delta Lake as our format, and
>> everything seemed great. But when running some one-off analytics we ran
>> into an issue. Specifically, we did something like:
>>
>> "df.<do some analytic>.repartition(1).write.csv()"
>>
>> This is our quick way of creating a CSV we can download and do other
>> things with when our result is some small aggregation. However, we kept
>> getting an empty output directory (just a _SUCCESS file and nothing else),
>> even though in the Spark UI it says it wrote some positive number of rows.
>> Eventually traced it back to our update to use the
>> ObserverReadProxyProvider in our notebook sessions. I finally figured out
>> it was due to the "Maintaining Client Consistency" section talked about in
>> https://hadoop.apache.org/docs/current/hadoop-project-dist/hadoop-hdfs/ObserverNameNode.html
>> .
>>
>> After setting the auto msync period to a low value, the writes started
>> working. I kept digging in and realized it's due to how the
>> FileOutputCommitter v1 algorithm works. During the commitJob phase, the
>> AM/driver does a file system listing on the output directory to find all
>> the finished task output files it needs to move to the top level output
>> directory. But since this is a read, the observer can serve this request,
>> but it can be out of date and not see the newly written files that just
>> finished from the executors. The auto msync fixed it because it forced the
>> driver to do an msync before the read took place. However, frequent auto
>> msyncs can defeat some of the performance benefits of the Observer.
>>
>> The v2 algorithm shouldn't have this issue because the tasks themselves
>> copy the output to the final directory when they finish, and the driver
>> simply adds the _SUCCESS at the end. And Hadoop's default is v2, but Spark
>> overrides that to use v1 by default because of potential correctness
>> issues, which is fair. While this is mostly an issue with Hadoop, the fact
>> that Spark defaults to the v1 algorithm makes it somewhat of a Spark
>> problem. Also, things like Delta Lake (or even regular structured streaming
>> output I think) shouldn't have issues because they are direct write with
>> transaction log based, so no file moving on the driver involved.
>>
>> So I mostly wanted to share that in case anyone else runs into this same
>> issue. But also wanted to get thoughts on if anything should be done about
>> this to prevent it from happening. Several ideas in no particular order:
>>
>> - Perform an msync during Spark's commitJob before calling the parent
>> commitJob. Since this is only available in newer APIs, probably isn't even
>> possible while maintaining compatibility with older Hadoop versions.
>> - Attempt to get an msync added upstream in Hadoop's v1 committer's
>> commitJob
>> - Attempt to detect the use of the ObserverReadProxyProvider and either
>> force using v2 committer on the spark side or just print out a warning that
>> you either need to use the v2 committer or you need to set the auto msync
>> period very low or 0 to guarantee correct output.
>> - Simply add something to the Spark docs somewhere about things to know
>> when using the ObserverReadProxyProvider
>> - Assume that if you are capable of creating your own Hadoop cluster with
>> an Observer Namenode you will recognize this limitation quickly, which it
>> only took me about an hour to figure out so that's also fair
>>
>> Thanks,
>>
>> --
>> Adam
>>
>

-- 
Adam Binford

Re: Observer Namenode and Committer Algorithm V1

[Venkatakrishnan Sowrirajan <vs...@asu.edu>]

I have created a JIRA (https://issues.apache.org/jira/browse/SPARK-36810)
to track this issue. Will look into this issue further in the coming days.

Regards
Venkata krishnan


On Tue, Sep 7, 2021 at 5:57 AM Steve Loughran <st...@cloudera.com.invalid>
wrote:

> FileContext came in Hadoop 2.x with a cleaner split of client API and
> driver implementation, and stricter definition of some things considered
> broken in FileSystem (rename() corner cases, notion of a current directory,
> ...)
>
> But as it came out after the platform was broadly adopted & never
> backported to hadoop 1, it never got picked up... So even though its tagged
> as the "newer" API, it's not the one used by apps. And as it will relay to
> FileSystem, anyone doing interesting things at the FS client level can just
> add it there and have it adopted in both places.
>
> The design of FileContext _is_ better, but the extra layers get in the way
> of the interesting games you can play to deliver performance speedups
> against cloud storage. So that's why we tend to work in FileSystem, with
> the FS API spec and contract tests essentially reverse engineering what it
> is that HDFS does and which applications expect (thread safety of input and
> output streams, rename() return codes, ...)
>
> FileSystem is never going to go away. I'd like to fix rename() but we
> can't change rename/2's semantics, making the protected rename/3 isn't
> sufficient. See https://github.com/apache/hadoop/pull/2735
> <https://urldefense.com/v3/__https://github.com/apache/hadoop/pull/2735__;!!IKRxdwAv5BmarQ!PNuufr0_QHextDKhP3kMaqTc2HJjvEcE50VXE4ah12vXr5C-1nIBGhBiEvgYh-Q$>
> for my lapsed work. Got too complicated for some spare-time work,
> especially when there are others with more tangible benefit which don't
> have good alternatives (https://github.com/apache/hadoop/pull/2584
> <https://urldefense.com/v3/__https://github.com/apache/hadoop/pull/2584__;!!IKRxdwAv5BmarQ!PNuufr0_QHextDKhP3kMaqTc2HJjvEcE50VXE4ah12vXr5C-1nIBGhBi1IJO5sU$>
> )
>
> On Mon, 6 Sept 2021 at 16:49, Adam Binford <ad...@gmail.com> wrote:
>
>> Sharing some things I learned looking into the Delta Lake issue:
>>
>> - This was a read after write inconsistency _all on the driver_.
>> Specifically it currently uses the FileSystem API for reading table logs
>> for greater compatibility, but the FileContext API for writes for atomic
>> renames. This led to the FileSystem reads becoming stale as they didn't
>> have to update their state ID after the FileContext writes from a different
>> DFSClient.
>> - The FileContext API generally seems less suitable for an HA HDFS setup,
>> as each FileContext object creates a new DFSClient that has to re-find the
>> active/observer nodes. I know these are cheap operations, but still extra
>> overhead and not ideal. This is compounded by the fact that the name
>> "FileContext" is misleading, as it sounds like something you should create
>> a new instance of per file you want to interact with, and not try to reuse
>> across a file system. There's been an open issue for 12 years about adding
>> caching to the FileContext/AbstractFileSystem API, but there seems some
>> hesitation there due to what happens when you update HDFS while a client is
>> still active. This doesn't appear to be a huge issue directly in spark,
>> since the main place FileContext is used is for the structured streaming
>> commit log for atomic renames, but something to look out for in third party
>> libraries. I do see a lot of warnings about the HDFSMetadataLog looking for
>> the active namenode which I haven't looked into much. I'd expect to only
>> see that once since it seems to properly reuse a single FileContext
>> instance.
>>
>> Adam
>>
>> On Fri, Aug 20, 2021 at 2:22 PM Steve Loughran
>> <st...@cloudera.com.invalid> wrote:
>>
>>>
>>> ooh, this is fun,
>>>
>>> v2 isn't safe to use unless every task attempt generates files with
>>> exactly the same names and it is okay to intermingle the output of two task
>>> attempts.
>>>
>>> This is because task commit can felt partway through (or worse, that
>>> process pause for a full GC), and a second attempt committed. Spark commit
>>> algorithm assumes that it's OK to commit a 2nd attempt if the first attempt
>>> failed, timed out etc. It is for v1, but not v2
>>>
>>> Therefore: a (nonbinding) -1 to any proposal to switch to v2. You are
>>> only changing problems
>>>
>>>
>>> I think the best fix here is to do it in the FileOutputCommitter. Be
>>> aware that we are all scared of that class and always want to do the
>>> minimum necessary.
>>>
>>> I will certainly add to the manifest committer, whose "call for
>>> reviewers and testing" is still open, especially all the way through spark
>>> https://github.com/apache/hadoop/pull/2971
>>> <https://urldefense.com/v3/__https://github.com/apache/hadoop/pull/2971__;!!IKRxdwAv5BmarQ!PNuufr0_QHextDKhP3kMaqTc2HJjvEcE50VXE4ah12vXr5C-1nIBGhBi-1KKzmM$>
>>>
>>> That committer works with HDFS too, I'd be interested in anyone
>>> benchmarking it on queries with deep/wide directory trees and with
>>> different tasks all generating output for the same destination directories
>>> (i.e file rename dominates in job commit, not task rename). I'm not
>>> optimising it for HDFS -it's trying to deal with cloud storage quirks like
>>> nonatomic dir rename (GCS), slow list/file rename perf (everywhere), deep
>>> directory delete timeouts, and other cloud storage specific issues.
>>>
>>>
>>> Further reading on the commit problem in general
>>> https://github.com/steveloughran/zero-rename-committer/releases/tag/tag_release_2021-05-17
>>> <https://urldefense.com/v3/__https://github.com/steveloughran/zero-rename-committer/releases/tag/tag_release_2021-05-17__;!!IKRxdwAv5BmarQ!PNuufr0_QHextDKhP3kMaqTc2HJjvEcE50VXE4ah12vXr5C-1nIBGhBiMBaygVI$>
>>>
>>> -Steve
>>>
>>>
>>>
>>> On Tue, 17 Aug 2021 at 17:39, Adam Binford <ad...@gmail.com> wrote:
>>>
>>>> Hi,
>>>>
>>>> We ran into an interesting issue that I wanted to share as well as get
>>>> thoughts on if anything should be done about this. We run our own Hadoop
>>>> cluster and recently deployed an Observer Namenode to take some burden off
>>>> of our Active Namenode. We mostly use Delta Lake as our format, and
>>>> everything seemed great. But when running some one-off analytics we ran
>>>> into an issue. Specifically, we did something like:
>>>>
>>>> "df.<do some analytic>.repartition(1).write.csv()"
>>>>
>>>> This is our quick way of creating a CSV we can download and do other
>>>> things with when our result is some small aggregation. However, we kept
>>>> getting an empty output directory (just a _SUCCESS file and nothing else),
>>>> even though in the Spark UI it says it wrote some positive number of rows.
>>>> Eventually traced it back to our update to use the
>>>> ObserverReadProxyProvider in our notebook sessions. I finally figured out
>>>> it was due to the "Maintaining Client Consistency" section talked about in
>>>> https://hadoop.apache.org/docs/current/hadoop-project-dist/hadoop-hdfs/ObserverNameNode.html
>>>> <https://urldefense.com/v3/__https://hadoop.apache.org/docs/current/hadoop-project-dist/hadoop-hdfs/ObserverNameNode.html__;!!IKRxdwAv5BmarQ!PNuufr0_QHextDKhP3kMaqTc2HJjvEcE50VXE4ah12vXr5C-1nIBGhBid87xBgQ$>
>>>> .
>>>>
>>>> After setting the auto msync period to a low value, the writes started
>>>> working. I kept digging in and realized it's due to how the
>>>> FileOutputCommitter v1 algorithm works. During the commitJob phase, the
>>>> AM/driver does a file system listing on the output directory to find all
>>>> the finished task output files it needs to move to the top level output
>>>> directory. But since this is a read, the observer can serve this request,
>>>> but it can be out of date and not see the newly written files that just
>>>> finished from the executors. The auto msync fixed it because it forced the
>>>> driver to do an msync before the read took place. However, frequent auto
>>>> msyncs can defeat some of the performance benefits of the Observer.
>>>>
>>>> The v2 algorithm shouldn't have this issue because the tasks themselves
>>>> copy the output to the final directory when they finish, and the driver
>>>> simply adds the _SUCCESS at the end. And Hadoop's default is v2, but Spark
>>>> overrides that to use v1 by default because of potential correctness
>>>> issues, which is fair. While this is mostly an issue with Hadoop, the fact
>>>> that Spark defaults to the v1 algorithm makes it somewhat of a Spark
>>>> problem. Also, things like Delta Lake (or even regular structured streaming
>>>> output I think) shouldn't have issues because they are direct write with
>>>> transaction log based, so no file moving on the driver involved.
>>>>
>>>> So I mostly wanted to share that in case anyone else runs into this
>>>> same issue. But also wanted to get thoughts on if anything should be done
>>>> about this to prevent it from happening. Several ideas in no particular
>>>> order:
>>>>
>>>> - Perform an msync during Spark's commitJob before calling the parent
>>>> commitJob. Since this is only available in newer APIs, probably isn't even
>>>> possible while maintaining compatibility with older Hadoop versions.
>>>> - Attempt to get an msync added upstream in Hadoop's v1 committer's
>>>> commitJob
>>>> - Attempt to detect the use of the ObserverReadProxyProvider and either
>>>> force using v2 committer on the spark side or just print out a warning that
>>>> you either need to use the v2 committer or you need to set the auto msync
>>>> period very low or 0 to guarantee correct output.
>>>> - Simply add something to the Spark docs somewhere about things to know
>>>> when using the ObserverReadProxyProvider
>>>> - Assume that if you are capable of creating your own Hadoop cluster
>>>> with an Observer Namenode you will recognize this limitation quickly, which
>>>> it only took me about an hour to figure out so that's also fair
>>>>
>>>> Thanks,
>>>>
>>>> --
>>>> Adam
>>>>
>>>
>>
>> --
>> Adam Binford
>>
>

Re: Observer Namenode and Committer Algorithm V1

[Steve Loughran <st...@cloudera.com.INVALID>]

FileContext came in Hadoop 2.x with a cleaner split of client API and
driver implementation, and stricter definition of some things considered
broken in FileSystem (rename() corner cases, notion of a current directory,
...)

But as it came out after the platform was broadly adopted & never
backported to hadoop 1, it never got picked up... So even though its tagged
as the "newer" API, it's not the one used by apps. And as it will relay to
FileSystem, anyone doing interesting things at the FS client level can just
add it there and have it adopted in both places.

The design of FileContext _is_ better, but the extra layers get in the way
of the interesting games you can play to deliver performance speedups
against cloud storage. So that's why we tend to work in FileSystem, with
the FS API spec and contract tests essentially reverse engineering what it
is that HDFS does and which applications expect (thread safety of input and
output streams, rename() return codes, ...)

FileSystem is never going to go away. I'd like to fix rename() but we can't
change rename/2's semantics, making the protected rename/3 isn't
sufficient. See https://github.com/apache/hadoop/pull/2735 for my lapsed
work. Got too complicated for some spare-time work, especially when there
are others with more tangible benefit which don't have good alternatives (
https://github.com/apache/hadoop/pull/2584)

On Mon, 6 Sept 2021 at 16:49, Adam Binford <ad...@gmail.com> wrote:

> Sharing some things I learned looking into the Delta Lake issue:
>
> - This was a read after write inconsistency _all on the driver_.
> Specifically it currently uses the FileSystem API for reading table logs
> for greater compatibility, but the FileContext API for writes for atomic
> renames. This led to the FileSystem reads becoming stale as they didn't
> have to update their state ID after the FileContext writes from a different
> DFSClient.
> - The FileContext API generally seems less suitable for an HA HDFS setup,
> as each FileContext object creates a new DFSClient that has to re-find the
> active/observer nodes. I know these are cheap operations, but still extra
> overhead and not ideal. This is compounded by the fact that the name
> "FileContext" is misleading, as it sounds like something you should create
> a new instance of per file you want to interact with, and not try to reuse
> across a file system. There's been an open issue for 12 years about adding
> caching to the FileContext/AbstractFileSystem API, but there seems some
> hesitation there due to what happens when you update HDFS while a client is
> still active. This doesn't appear to be a huge issue directly in spark,
> since the main place FileContext is used is for the structured streaming
> commit log for atomic renames, but something to look out for in third party
> libraries. I do see a lot of warnings about the HDFSMetadataLog looking for
> the active namenode which I haven't looked into much. I'd expect to only
> see that once since it seems to properly reuse a single FileContext
> instance.
>
> Adam
>
> On Fri, Aug 20, 2021 at 2:22 PM Steve Loughran <st...@cloudera.com.invalid>
> wrote:
>
>>
>> ooh, this is fun,
>>
>> v2 isn't safe to use unless every task attempt generates files with
>> exactly the same names and it is okay to intermingle the output of two task
>> attempts.
>>
>> This is because task commit can felt partway through (or worse, that
>> process pause for a full GC), and a second attempt committed. Spark commit
>> algorithm assumes that it's OK to commit a 2nd attempt if the first attempt
>> failed, timed out etc. It is for v1, but not v2
>>
>> Therefore: a (nonbinding) -1 to any proposal to switch to v2. You are
>> only changing problems
>>
>>
>> I think the best fix here is to do it in the FileOutputCommitter. Be
>> aware that we are all scared of that class and always want to do the
>> minimum necessary.
>>
>> I will certainly add to the manifest committer, whose "call for reviewers
>> and testing" is still open, especially all the way through spark
>> https://github.com/apache/hadoop/pull/2971
>>
>> That committer works with HDFS too, I'd be interested in anyone
>> benchmarking it on queries with deep/wide directory trees and with
>> different tasks all generating output for the same destination directories
>> (i.e file rename dominates in job commit, not task rename). I'm not
>> optimising it for HDFS -it's trying to deal with cloud storage quirks like
>> nonatomic dir rename (GCS), slow list/file rename perf (everywhere), deep
>> directory delete timeouts, and other cloud storage specific issues.
>>
>>
>> Further reading on the commit problem in general
>> https://github.com/steveloughran/zero-rename-committer/releases/tag/tag_release_2021-05-17
>>
>> -Steve
>>
>>
>>
>> On Tue, 17 Aug 2021 at 17:39, Adam Binford <ad...@gmail.com> wrote:
>>
>>> Hi,
>>>
>>> We ran into an interesting issue that I wanted to share as well as get
>>> thoughts on if anything should be done about this. We run our own Hadoop
>>> cluster and recently deployed an Observer Namenode to take some burden off
>>> of our Active Namenode. We mostly use Delta Lake as our format, and
>>> everything seemed great. But when running some one-off analytics we ran
>>> into an issue. Specifically, we did something like:
>>>
>>> "df.<do some analytic>.repartition(1).write.csv()"
>>>
>>> This is our quick way of creating a CSV we can download and do other
>>> things with when our result is some small aggregation. However, we kept
>>> getting an empty output directory (just a _SUCCESS file and nothing else),
>>> even though in the Spark UI it says it wrote some positive number of rows.
>>> Eventually traced it back to our update to use the
>>> ObserverReadProxyProvider in our notebook sessions. I finally figured out
>>> it was due to the "Maintaining Client Consistency" section talked about in
>>> https://hadoop.apache.org/docs/current/hadoop-project-dist/hadoop-hdfs/ObserverNameNode.html
>>> .
>>>
>>> After setting the auto msync period to a low value, the writes started
>>> working. I kept digging in and realized it's due to how the
>>> FileOutputCommitter v1 algorithm works. During the commitJob phase, the
>>> AM/driver does a file system listing on the output directory to find all
>>> the finished task output files it needs to move to the top level output
>>> directory. But since this is a read, the observer can serve this request,
>>> but it can be out of date and not see the newly written files that just
>>> finished from the executors. The auto msync fixed it because it forced the
>>> driver to do an msync before the read took place. However, frequent auto
>>> msyncs can defeat some of the performance benefits of the Observer.
>>>
>>> The v2 algorithm shouldn't have this issue because the tasks themselves
>>> copy the output to the final directory when they finish, and the driver
>>> simply adds the _SUCCESS at the end. And Hadoop's default is v2, but Spark
>>> overrides that to use v1 by default because of potential correctness
>>> issues, which is fair. While this is mostly an issue with Hadoop, the fact
>>> that Spark defaults to the v1 algorithm makes it somewhat of a Spark
>>> problem. Also, things like Delta Lake (or even regular structured streaming
>>> output I think) shouldn't have issues because they are direct write with
>>> transaction log based, so no file moving on the driver involved.
>>>
>>> So I mostly wanted to share that in case anyone else runs into this same
>>> issue. But also wanted to get thoughts on if anything should be done about
>>> this to prevent it from happening. Several ideas in no particular order:
>>>
>>> - Perform an msync during Spark's commitJob before calling the parent
>>> commitJob. Since this is only available in newer APIs, probably isn't even
>>> possible while maintaining compatibility with older Hadoop versions.
>>> - Attempt to get an msync added upstream in Hadoop's v1 committer's
>>> commitJob
>>> - Attempt to detect the use of the ObserverReadProxyProvider and either
>>> force using v2 committer on the spark side or just print out a warning that
>>> you either need to use the v2 committer or you need to set the auto msync
>>> period very low or 0 to guarantee correct output.
>>> - Simply add something to the Spark docs somewhere about things to know
>>> when using the ObserverReadProxyProvider
>>> - Assume that if you are capable of creating your own Hadoop cluster
>>> with an Observer Namenode you will recognize this limitation quickly, which
>>> it only took me about an hour to figure out so that's also fair
>>>
>>> Thanks,
>>>
>>> --
>>> Adam
>>>
>>
>
> --
> Adam Binford
>