venkata91 commented on code in PR #17820: URL: https://github.com/apache/iceberg/pull/17820#discussion_r4533574585
########## spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/WriteDurations.java: ########## @@ -0,0 +1,58 @@ +/* + * Licensed to the Apache Software Foundation (ASF) under one + * or more contributor license agreements. See the NOTICE file + * distributed with this work for additional information + * regarding copyright ownership. The ASF licenses this file + * to you under the Apache License, Version 2.0 (the + * "License"); you may not use this file except in compliance + * with the License. You may obtain a copy of the License at + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, + * software distributed under the License is distributed on an + * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY + * KIND, either express or implied. See the License for the + * specific language governing permissions and limitations + * under the License. + */ +package org.apache.iceberg.spark; + +import java.io.Serializable; + +/** + * Write times summed across the tasks of a write. + * + * <p>Tasks report their own times in their commit messages. Spark collects task metrics before it + * calls {@code DataWriter.commit()}, so the close time, where buffered rows are flushed, is not yet + * known at that point and cannot be reported as a task metric. Both times therefore travel back to + * the driver with the commit messages and are aggregated here. + */ +public class WriteDurations implements Serializable { Review Comment: Done in e828924. `WriteDurations` is now package-private in `org.apache.iceberg.spark.source`, and `customTaskMetrics` takes the write and close times as two longs. ########## spark/v4.1/spark/src/main/java/org/apache/iceberg/spark/source/SparkWrite.java: ########## @@ -285,20 +287,24 @@ private void abort(WriterCommitMessage[] messages) { private DataFileSet files(WriterCommitMessage[] messages) { DataFileSet files = DataFileSet.create(); + // every commit path funnels through here, so this is where task write times are collected + WriteDurations durations = WriteDurations.EMPTY; for (WriterCommitMessage message : messages) { if (message != null) { TaskCommit taskCommit = (TaskCommit) message; files.addAll(Arrays.asList(taskCommit.files())); + durations = durations.combine(taskCommit.writeDurations()); Review Comment: Good catch, min/med/max on a driver-side value just repeat the total, which isn't useful. I looked at moving it to `currentMetricsValues()`. In Spark 4.1, `WritingSparkTask` collects the task metrics once, right before it calls `DataWriter.commit()`, and our `commit()` is where the writer gets closed and the buffered rows are flushed. So there are basically two options: 1. Keep it on the driver (what the PR does now). The write and close totals are both accurate, but min/med/max show the total. 2. Report `writeDuration` from the tasks through `currentMetricsValues()`. That gives a real min/med/max, but only for the row-level write calls. The close time can't be a task metric, so it still goes through the commit messages with only a total. For files smaller than a row group I think most of the actual I/O happens in that close, so the per-task numbers could understate the slow tasks. Neither is great. What do you think, would you prefer one of these, or is there another way you'd go about it? -- This is an automated message from the Apache Git Service. To respond to the message, please log on to GitHub and use the URL above to go to the specific comment. To unsubscribe, e-mail: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
