Hi Robert, Any updates on the below for the community?
Thanks, M On Tue, Apr 25, 2017 at 8:50 AM, Robert Schmidtke <ro.schmid...@gmail.com> wrote: > Hi Ufuk, thanks for coming back to me on this. > > The records are 100 bytes in size, the benchmark being TeraSort, so that > should not be an issue. I have played around with the input size, and here > are my observations: > > 128 GiB input: 0 Spilling in Flink. > 256 GiB input: 88 GiB Spilling in Flink (so 88 GiB of reads, 88 GiB of > writes), and my instrumentation covers all of it. > 384 GiB input: 391 GiB Spilling in Flink, and I cover all of it. > 512 GiB input: 522 GiB Spilling in Flink, but I miss 140 GiB of it. > 640 GiB input: 653 GiB Spilling in Flink, but I miss 281 GiB of it. > 768 GiB input: 784 GiB Spilling in Flink, but I miss 490 GiB of it. > 896 GiB input: 914 GiB Spilling in Flink, but I miss 662 GiB of it. > 1024 GiB input: 1045 GiB Spilling in Flink, but I miss 968 GiB of it. > > So regardless of how well configured my system is and spilling is even > necessary, it seems that with larger spilling amounts, the way the data is > spilled changes (and I start missing larger and larger portions of I/O > until almost 100%). > Now since I have written the instrumentation myself, I cannot guarantee > that it is flawless and I might have missed something. > I'm currently looking into how the file channels are being accessed in > parallel by multiple threads, which I cover as well and my tests verify it, > but maybe there are special access patterns here. > > Robert > > On Mon, Apr 24, 2017 at 2:25 PM, Ufuk Celebi <u...@apache.org> wrote: > >> Hey Robert, >> >> for batch that should cover the relevant spilling code. If the records >> are >= 5 MB, the SpillingAdaptiveSpanningRecordDeserializer will spill >> incoming records as well. But that should be covered by the >> FileChannel instrumentation as well? >> >> – Ufuk >> >> >> On Tue, Apr 18, 2017 at 3:57 PM, Robert Schmidtke >> <ro.schmid...@gmail.com> wrote: >> > Hi, >> > >> > I have already looked at the UnilateralSortMerger, concluding that all >> I/O >> > eventually goes via SegmentReadRequest and SegmentWriteRequest (which in >> > turn use java.nio.channels.FileChannel) in AsynchronousFileIOChannel. >> Are >> > there more interaction points between Flink and the underlying file >> system >> > that I might want to consider? >> > >> > Thanks! >> > Robert >> > >> > On Fri, Apr 7, 2017 at 5:02 PM, Kurt Young <ykt...@gmail.com> wrote: >> >> >> >> Hi, >> >> >> >> You probably want check out UnilateralSortMerger.java, this is the >> class >> >> which is responsible for external sort for flink. Here is a short >> >> description for how it works: there are totally 3 threads working >> together, >> >> one for reading, one for sorting partial data in memory, and the last >> one is >> >> responsible for spilling. Flink will first figure out how many memory >> it can >> >> use during the in-memory sort, and manage them as MemorySegments. Once >> these >> >> memory runs out, the sorting thread will take over these memory and do >> the >> >> in-memory sorting (For more details about in-memory sorting, you can >> see >> >> NormalizedKeySorter). After this, the spilling thread will write this >> sorted >> >> data to disk and make these memory available again for reading. This >> will >> >> repeated until all data has been processed. >> >> Normally, the data will be read twice (one from source, and one from >> disk) >> >> and write once, but if you spilled too much files, flink will first >> merge >> >> some all the files and make sure the last merge step will not exceed >> some >> >> limit (default 128). Hope this can help you. >> >> >> >> Best, >> >> Kurt >> >> >> >> On Fri, Apr 7, 2017 at 4:20 PM, Robert Schmidtke < >> ro.schmid...@gmail.com> >> >> wrote: >> >>> >> >>> Hi, >> >>> >> >>> I'm currently examining the I/O patterns of Flink, and I'd like to >> know >> >>> when/how Flink goes to disk. Let me give an introduction of what I >> have done >> >>> so far. >> >>> >> >>> I am running TeraGen (from the Hadoop examples package) + TeraSort >> >>> (https://github.com/robert-schmidtke/terasort) on a 16 node cluster, >> each >> >>> node with 64 GiB of memory, 2x32 cores, and roughly half a terabyte >> of disk. >> >>> I'm using YARN and HDFS. The underlying file system is XFS. >> >>> >> >>> Now before running TeraGen and TeraSort, I reset the XFS counters to >> >>> zero, and after TeraGen + TeraSort are finished, I dump the XFS >> counters >> >>> again. Accumulated over the entire cluster I get 3 TiB of writes and >> 3.2 TiB >> >>> of reads. What I'd have expected would be 2 TiB of writes (1 for >> TeraGen, 1 >> >>> for TeraSort) and 1 TiB of reads (during TeraSort). >> >>> >> >>> Unsatisfied by the coarseness of these numbers I developed an HDFS >> >>> wrapper that logs file system statistics for each call to hdfs://..., >> such >> >>> as start time/end time, no. of bytes read/written etc. I can plot >> these >> >>> numbers and see what I expect: during TeraGen I have 1 TiB of writes >> to >> >>> hdfs://..., during TeraSort I have 1 TiB of reads from and 1 TiB of >> writes >> >>> to hdfs://... So far, so good. >> >>> >> >>> Now this still did not explain the disk I/O, so I added bytecode >> >>> instrumentation to a range of Java classes, like FileIn/OutputStream, >> >>> RandomAccessFile, FileChannel, ZipFile, multiple *Buffer classes for >> memory >> >>> mapped files etc., and have the same statistics: start/end of a read >> >>> from/write to disk, no. of bytes involved and such. I can plot these >> numbers >> >>> too and see that the HDFS JVMs write 1 TiB of data to disk during >> TeraGen >> >>> (expected) and read and write 1 TiB from and to disk during TeraSort >> >>> (expected). >> >>> >> >>> Sorry for the enormous introduction, but now there's finally the >> >>> interesting part: Flink's JVMs read from and write to disk 1 TiB of >> data >> >>> each during TeraSort. I'm suspecting there is some sort of spilling >> >>> involved, potentially because I have not done the setup properly. But >> that >> >>> is not the crucial point: my statistics give a total of 3 TiB of >> writes to >> >>> disk (2 TiB for HDFS, 1 TiB for Flink), which agrees with the XFS >> counters >> >>> from above. However, my statistics only give 2 TiB of reads from disk >> (1 TiB >> >>> for HDFS, 1 TiB for Flink), so I'm missing an entire TiB of reads >> from disk >> >>> somewhere. I have done the same with Hadoop TeraSort, and there I'm >> not >> >>> missing any data, meaning my statistics agree with XFS for TeraSort on >> >>> Hadoop, which is why I suspect there are some cases where Flink goes >> to disk >> >>> without me noticing it. >> >>> >> >>> Therefore here finally the question: in which cases does Flink go to >> >>> disk, and how does it do so (meaning precisely which Java classes are >> >>> involved, so I can check my bytecode instrumentation)? This would also >> >>> include any kind of resource distribution via HDFS/YARN I guess (like >> JAR >> >>> files and I don't know what). Seeing that I'm missing an amount of >> data >> >>> equal to the size of my input set I'd suspect there must be some sort >> of >> >>> shuffling/spilling at play here, but I'm not sure. Maybe there is >> also some >> >>> sort of remote I/O involved via sockets or so that I'm missing. >> >>> >> >>> Any hints as to where Flink might incur disk I/O are greatly >> appreciated! >> >>> I'm also happy with doing the digging myself, once pointed to the >> proper >> >>> packages in the Apache Flink source tree (I have done my fair share of >> >>> inspection already, but could not be sure whether or not I have missed >> >>> something). Thanks a lot in advance! >> >>> >> >>> Robert >> >>> >> >>> -- >> >>> My GPG Key ID: 336E2680 >> >> >> >> >> > >> > >> > >> > -- >> > My GPG Key ID: 336E2680 >> > > > > -- > My GPG Key ID: 336E2680 >
