Skip to content
Closed
Show file tree
Hide file tree
Changes from 5 commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -0,0 +1,61 @@
/*
* 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.spark.sql.connector;

import org.apache.spark.annotation.Evolving;
import org.apache.spark.sql.connector.read.PartitionReader;
import org.apache.spark.sql.connector.read.Scan;

/**
* A custom metric. This is a logical representation of a metric reported by data sources during
* read path. Data sources can report supported metric list by {@link Scan} to Spark in query
* planning. During query execution, Spark will collect the metrics per partition by
* {@link PartitionReader} and combine metrics from partitions to the final result. How Spark
* combines metrics depends on the metric type. For streaming query, Spark will collect and combine
* metrics for a final result per micro batch.
*
* The metrics will be gathered during query execution back to the driver and then combined. The
* final result will be shown up in the physical operator in Spark UI.
*
* @since 3.2.0
*/
@Evolving
public interface CustomMetric {
/**
* Returns the name of custom metric.
*/
String name();

/**
* Returns the description of custom metric.
*/
String description();

/**
* Supported metric type. The metric types must be supported by Spark SQL internal metrics.
Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Spark SQL internal metrics always return a long value. Do we still need the LongMetric interface if metrics types are limited to SQL internal metrics?

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

For now a CustomMetric which returns long value should be enough. Previously there is a suggestion to name it just LongMetric because it always returns long value. But it sounds too concrete.

So I think is it good to leave some flexibility here? Then I extract LongMetric out as a separate interface which returns long value.

I'm fine to just return long in CustomMetric (and maybe name it to LongMetric).

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'll be +1 on returning the long value directly in CustomMetric. I don't have a strong opinion about the naming.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Okay, let me move it back. See if others have different thoughts.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

The metric types must be supported by Spark SQL internal metrics.

It sounds weird that we require users to understand internal private APIs when using a public API.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

"The metric types must be supported by Spark SQL internal metrics." is more for the developers trying to add new metrics here, instead of using the API. I can remove it if it sounds inappropriate.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

+1 for removing it. I think users just need to know the below line: SUM: Spark sums up metrics from partitions as the final result..

* SUM: Spark sums up metrics from partitions as the final result.
*/
enum MetricType {
SUM
Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I only use sum metric for now for the Kafka scan purpose. So I leave other possible metric type (size, timing, average) out for now to make it simpler at the beginning.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Any reason why we do not allow users to define the combine method instead? The current API means if a user needs to use a different combine behavior, they need to submit a PR to Spark to add it and wait for a new Spark release.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

By the way, regarding the naming, I would prefer MergeType to make it clear if we did go to this direction.

Copy link
Member Author

@viirya viirya Feb 22, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Because these metrics are logical representation and are collected to be SQL metrics internally. They are used by DS v2 to report SQL metrics easily without dealing with internal SQLMetric. So SQL metrics define how metrics are combined/aggregated. A similar case is public expression APIs for predicate pushdown, as they are converted from catalyst expressions so are matched to catalyst expressions.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I'm wondering if we can design a better API. This would be a public API and hard to change in future. We don't have to build it in a way just because of what SQLMetric can do today. When I added SQLMetric to Spark SQL, we didn't pay much attention to this since it's an internal API. But making a new public API is a different story and we need to think about how to use it from the user perspective.

Copy link
Member Author

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Hmm, these metrics are not general metrics that could be defined and used by end users. It sounds like we already have Accumulator for the purpose. Defining a public API of general metrics for end users is overlapping with Accumulator, IMHO.

The metrics here are used by DS v2 implementations to report metrics to Spark SQL. That being said, it is not exposed for end users as general metrics. I think the purpose for the metric API is clear.

Copy link
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Defining a public API of general metrics for end users is overlapping with Accumulator, IMHO.

Totally agree. There would be 3 types of metrics APIs (including the internal SQLMetric) if we added this. That's really confusing to users. Is it possible to make SQLMetric support general Accmulator instead so that we don't need to re-invent a new metric API?

That being said, it is not exposed for end users as general metrics.

Developers building spark data sources are also critical to the whole Spark ecosystem and these APIs are designed for them. IMHO, we should try our best to build a better API if possible.

Copy link
Member Author

@viirya viirya Feb 23, 2021

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

As you can see, the metric API here is a logical representation of metrics from DS v2. We are not going to re-invent a whole metric API. SQLMetrics are internal to Spark. It is not exposed to end users and data source developers, so I don't think it worries me too much.

I'm not saying that we should not build a good API for DS v2 developers. Seems to me some points in above comments are from end user perspective, I'd like to point out this is for different scenarios.

As this is used for DS v2 purpose, it is for SQL metrics and internally it is converted to SQL metrics. To make SQLMetric support Accmulator and let DS v2 reports Accmulator does not sound bad idea to me. But I'd doubt if it is worth.

One argued point is to define arbitrary combine behavior. Once making SQLMetric support Accmulator, does it mean that we can use arbitrary Accmulator? No, basically SQLMetric allows certain types of metrics, we still need to change SQLMetric to support new metrics.

So the only benefit I thought is to not have another metric API. And I don't think it is serious for this case at the beginning.

This API is pretty simple as it is just logical representation and we only need small change internally to convert collected metrics from DS v2 to SQL metrics.

I just read through the code path to be touched in order to make SQLMetric support Accmulator. Seems it involves more changes not only in DS v2 but maybe also other stuff in sql/core, etc.

Although I doubt if it is worth, I'm open to the suggested Accmulator approach. Let's gather more thoughts from others?

cc @cloud-fan @dongjoon-hyun @rdblue @Ngone51 @sunchao WDYT?

}

/**
* Returns the type of custom metric.
*/
MetricType type();
}
Original file line number Diff line number Diff line change
@@ -0,0 +1,33 @@
/*
* 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.spark.sql.connector;

import org.apache.spark.annotation.Evolving;

/**
* A custom metric that reports a long value.
*
* @since 3.2.0
*/
@Evolving
public interface LongMetric extends CustomMetric {
/**
* Returns the value of custom metric.
*/
long value();
}
Original file line number Diff line number Diff line change
Expand Up @@ -21,6 +21,7 @@
import java.io.IOException;

import org.apache.spark.annotation.Evolving;
import org.apache.spark.sql.connector.CustomMetric;

/**
* A partition reader returned by {@link PartitionReaderFactory#createReader(InputPartition)} or
Expand Down Expand Up @@ -48,4 +49,12 @@ public interface PartitionReader<T> extends Closeable {
* Return the current record. This method should return same value until `next` is called.
*/
T get();

/**
* Returns an array of custom metrics. By default it returns empty array.
*/
default CustomMetric[] getCustomMetrics() {
CustomMetric[] NO_METRICS = {};
return NO_METRICS;
}
}
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
package org.apache.spark.sql.connector.read;

import org.apache.spark.annotation.Evolving;
import org.apache.spark.sql.connector.CustomMetric;
import org.apache.spark.sql.connector.read.streaming.ContinuousStream;
import org.apache.spark.sql.connector.read.streaming.MicroBatchStream;
import org.apache.spark.sql.types.StructType;
Expand Down Expand Up @@ -102,4 +103,13 @@ default MicroBatchStream toMicroBatchStream(String checkpointLocation) {
default ContinuousStream toContinuousStream(String checkpointLocation) {
throw new UnsupportedOperationException(description() + ": Continuous scan are not supported");
}

/**
* Returns an array of supported custom metrics with name and description.
* By default it returns empty array.
*/
default CustomMetric[] supportedCustomMetrics() {
CustomMetric[] NO_METRICS = {};
return NO_METRICS;
}
}