huaxingao commented on code in PR #17820: URL: https://github.com/apache/iceberg/pull/17820#discussion_r4527645797
########## 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: Does this need to be public? It's only used by the write classes in `org.apache.iceberg.spark.source` and by `SparkWriteUtil.customTaskMetrics`. If `customTaskMetrics` took the two long values instead, this class could move to the source package next to `WriteTimer` and be package-private. ########## 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: The task times are summed here and reported from the driver as one value, so the UI will show the same value for total, min, med and max. For example, tasks of 1 ms, 1 ms and 8 ms would show 10 ms (10 ms, 10 ms, 10 ms) instead of 10 ms (1 ms, 1 ms, 8 ms) for total (min, med, max). Could writeDuration be reported as a task metric through `currentMetricsValues()`, like `scanDuration`? -- 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]
