聊聊flink Table的Time Attributes
摘要：
序
本文主要研究一下flink Table的Time Attributes
Processing time
通過fromDataStream定義
DataStream<Tuple2<String, String>> stream = ...;
// d...
序
本文主要研究一下flink Table的Time Attributes
Processing time
通過fromDataStream定義
DataStream<Tuple2<String, String>> stream = ...;
// declare an additional logical field as a processing time attribute
Table table = tEnv.fromDataStream(stream, "Username, Data, UserActionTime.proctime");
WindowedTable windowedTable = table.window(Tumble.over("10.minutes").on("UserActionTime").as("userActionWindow"));
- 從DataStream建立Table的話，可以在fromDataStream裡頭進行定義Processing time
通過TableSource定義
// define a table source with a processing attribute
public class UserActionSource implements StreamTableSource<Row>, DefinedProctimeAttribute {
@Override
public TypeInformation<Row> getReturnType() {
String[] names = new String[] {"Username" , "Data"};
TypeInformation[] types = new TypeInformation[] {Types.STRING(), Types.STRING()};
return Types.ROW(names, types);
}
@Override
public DataStream<Row> getDataStream(StreamExecutionEnvironment execEnv) {
// create stream
DataStream<Row> stream = ...;
return stream;
}
@Override
public String getProctimeAttribute() {
// field with this name will be appended as a third field
return "UserActionTime";
}
}
// register table source
tEnv.registerTableSource("UserActions", new UserActionSource());
WindowedTable windowedTable = tEnv
.scan("UserActions")
.window(Tumble.over("10.minutes").on("UserActionTime").as("userActionWindow"));
- 通過TableSource建立Table的話，可以通過實現DefinedProctimeAttribute介面來定義Processing time
Event time
通過fromDataStream定義
// Option 1:
// extract timestamp and assign watermarks based on knowledge of the stream
DataStream<Tuple2<String, String>> stream = inputStream.assignTimestampsAndWatermarks(...);
// declare an additional logical field as an event time attribute
Table table = tEnv.fromDataStream(stream, "Username, Data, UserActionTime.rowtime");
// Option 2:
// extract timestamp from first field, and assign watermarks based on knowledge of the stream
DataStream<Tuple3<Long, String, String>> stream = inputStream.assignTimestampsAndWatermarks(...);
// the first field has been used for timestamp extraction, and is no longer necessary
// replace first field with a logical event time attribute
Table table = tEnv.fromDataStream(stream, "UserActionTime.rowtime, Username, Data");
// Usage:
WindowedTable windowedTable = table.window(Tumble.over("10.minutes").on("UserActionTime").as("userActionWindow"));
- 從DataStream建立Table的話，可以在fromDataStream裡頭進行定義Event time；具體有兩種方式，一種是額外定義一個欄位，一種是覆蓋原有的欄位
通過TableSource定義
// define a table source with a rowtime attribute
public class UserActionSource implements StreamTableSource<Row>, DefinedRowtimeAttributes {
@Override
public TypeInformation<Row> getReturnType() {
String[] names = new String[] {"Username", "Data", "UserActionTime"};
TypeInformation[] types =
new TypeInformation[] {Types.STRING(), Types.STRING(), Types.LONG()};
return Types.ROW(names, types);
}
@Override
public DataStream<Row> getDataStream(StreamExecutionEnvironment execEnv) {
// create stream
// ...
// assign watermarks based on the "UserActionTime" attribute
DataStream<Row> stream = inputStream.assignTimestampsAndWatermarks(...);
return stream;
}
@Override
public List<RowtimeAttributeDescriptor> getRowtimeAttributeDescriptors() {
// Mark the "UserActionTime" attribute as event-time attribute.
// We create one attribute descriptor of "UserActionTime".
RowtimeAttributeDescriptor rowtimeAttrDescr = new RowtimeAttributeDescriptor(
"UserActionTime",
new ExistingField("UserActionTime"),
new AscendingTimestamps());
List<RowtimeAttributeDescriptor> listRowtimeAttrDescr = Collections.singletonList(rowtimeAttrDescr);
return listRowtimeAttrDescr;
}
}
// register the table source
tEnv.registerTableSource("UserActions", new UserActionSource());
WindowedTable windowedTable = tEnv
.scan("UserActions")
.window(Tumble.over("10.minutes").on("UserActionTime").as("userActionWindow"));
- 通過TableSource建立Table的話，可以通過實現DefinedRowtimeAttributes介面來定義Event time
definedTimeAttributes
flink-table_2.11-1.7.0-sources.jar!/org/apache/flink/table/sources/definedTimeAttributes.scala
/**
* Extends a [[TableSource]] to specify a processing time attribute.
*/
trait DefinedProctimeAttribute {
/**
* Returns the name of a processing time attribute or null if no processing time attribute is
* present.
*
* The referenced attribute must be present in the [[TableSchema]] of the [[TableSource]] and of
* type [[Types.SQL_TIMESTAMP]].
*/
@Nullable
def getProctimeAttribute: String
}
/**
* Extends a [[TableSource]] to specify rowtime attributes via a
* [[RowtimeAttributeDescriptor]].
*/
trait DefinedRowtimeAttributes {
/**
* Returns a list of [[RowtimeAttributeDescriptor]] for all rowtime attributes of the table.
*
* All referenced attributes must be present in the [[TableSchema]] of the [[TableSource]] and of
* type [[Types.SQL_TIMESTAMP]].
*
* @return A list of [[RowtimeAttributeDescriptor]].
*/
def getRowtimeAttributeDescriptors: util.List[RowtimeAttributeDescriptor]
}
/**
* Describes a rowtime attribute of a [[TableSource]].
*
* @param attributeName The name of the rowtime attribute.
* @param timestampExtractor The timestamp extractor to derive the values of the attribute.
* @param watermarkStrategy The watermark strategy associated with the attribute.
*/
class RowtimeAttributeDescriptor(
val attributeName: String,
val timestampExtractor: TimestampExtractor,
val watermarkStrategy: WatermarkStrategy) {
/** Returns the name of the rowtime attribute. */
def getAttributeName: String = attributeName
/** Returns the [[TimestampExtractor]] for the attribute. */
def getTimestampExtractor: TimestampExtractor = timestampExtractor
/** Returns the [[WatermarkStrategy]] for the attribute. */
def getWatermarkStrategy: WatermarkStrategy = watermarkStrategy
override def equals(other: Any): Boolean = other match {
case that: RowtimeAttributeDescriptor =>
Objects.equals(attributeName, that.attributeName) &&
Objects.equals(timestampExtractor, that.timestampExtractor) &&
Objects.equals(watermarkStrategy, that.watermarkStrategy)
case _ => false
}
override def hashCode(): Int = {
Objects.hash(attributeName, timestampExtractor, watermarkStrategy)
}
}
- DefinedProctimeAttribute定義了getProctimeAttribute方法，返回String，用於定義Process time的欄位名；DefinedRowtimeAttributes定義了getRowtimeAttributeDescriptors方法，返回的是RowtimeAttributeDescriptor的List，RowtimeAttributeDescriptor有3個屬性，分別是attributeName、timestampExtractor及watermarkStrategy
小結
- 在從DataStream或者TableSource建立Table時可以指定Time Attributes，指定了之後就可以作為field來使用或者參與time-based的操作
- 針對Processing time，如果從DataStream建立Table的話，可以在fromDataStream裡頭進行定義；通過TableSource建立Table的話，可以通過實現DefinedProctimeAttribute介面來定義Processing time；DefinedProctimeAttribute定義了getProctimeAttribute方法，返回String，用於定義Process time的欄位名
- 針對Event time，如果從DataStream建立Table的話，可以在fromDataStream裡頭進行定義；具體有兩種方式，一種是額外定義一個欄位，一種是覆蓋原有的欄位；通過TableSource建立Table的話，可以通過實現DefinedRowtimeAttributes介面來定義Event time；DefinedRowtimeAttributes定義了getRowtimeAttributeDescriptors方法，返回的是RowtimeAttributeDescriptor的List，RowtimeAttributeDescriptor有3個屬性，分別是attributeName、timestampExtractor及watermarkStrategy