在流计算中,时间属性承担了一个极其重要的作用,所有基于时间的操作,例如窗口操作,都需要正确获取时间信息。我们曾经在Flink 源码阅读笔记(12)- 时间、定时器和窗口这篇文章中分析过 Flink 内部时间属性、水位线等机制的具体实现。在这篇文章中,我们将介绍在 SQL 和 Table API 中时间属性相关的一些细节。

时间属性概览

在 Flink SQL 中,表可以提供逻辑上的时间属性用于获取时间信息,时间属性可以是处理时间也可以是事件时间。在声明一张表的时候,时间属性可以在表的 schema 中定义。有些特定的操作,如窗口关联和窗口聚合操作必须基于时间属性字段,因而时间属性可以被看作一种特殊的字段类型;但是时间属性可以当作常规的时间戳字段来使用,一旦需要在计算中使用到时间属性,就需要“物化”(materialized)时间属性,时间属性字段就会被转换成一个常规的时间戳类型。被物化后的时间属性不再与 Flink 的时间系统和水位线相关联,因而也就不可以再应用在基于时间的操作中。

在 Flink SQL 的类型系统中,时间属性和常规的时间戳类型共用同样的逻辑类型 TimestampType,但是通过 TimestampKind 进行区分:

public enum TimestampKind {
	REGULAR, //常规的时间戳类型
	ROWTIME, //事件时间
	PROCTIME //处理时间
}

由于 Flink SQL 使用 Calcite 完成查询计划的优化,Flink 的所有逻辑类型在 Calcite 中都有对应的 RelDataType,并且为时间属性单独创建了一种新的 RelDataType,即 TimeIndicatorRelDataType:

class TimeIndicatorRelDataType(
    typeSystem: RelDataTypeSystem,
    originalType: BasicSqlType,
    val isEventTime: Boolean) //通过 isEventTime 区分是事件时间还是处理时间
  extends BasicSqlType(
    typeSystem,
    originalType.getSqlTypeName,
    originalType.getPrecision) {
}

定义表的时间属性

有两种方式来定义一张表的时间属性,一种方式是在将 DataStream 转换成 Table 的过程中,另一种方式是直接在 TableSource 的具体实现中定义。

DataStream 转换为 Table

在将 DataStream 转换成一个 Table 的过程中,可以用特殊的表达式来声明时间属性对应的列:
//处理时间
val table = tEnv.fromDataStream(stream, 'UserActionTimestamp, 'Username, 'Data, 'UserActionTime.proctime)

//事件时间
val table = tEnv.fromDataStream(stream, 'Username, 'Data, 'UserActionTime.rowtime)

其中 'field.proctime 或这 'field.rowtime 即声明时间属性,表达式会被转换为 PROCTIME 和 ROWTIME 这两个内置函数的调用 UnresolvedCallExpression

  // Time definition
  /**
    * Declares a field as the rowtime attribute for indicating, accessing, and working in
    * Flink's event time.
    */
  def rowtime: Expression = unresolvedCall(ROWTIME, expr)

  /**
    * Declares a field as the proctime attribute for indicating, accessing, and working in
    * Flink's processing time.
    */
  def proctime: Expression = unresolvedCall(PROCTIME, expr)

非 Scala 环境下则通过 ExpressionParser 完成表达式的解析。

使用 TableSource

如果要在 TableSource 中定义时间属性,则需要 TableSource 实现 DefinedProctimeAttribute 或者 DefinedRowAttribute 接口,并且引用的时间属性必须出现在 TableSchema 中,类型为 timestamp 类型。如果要同时使用处理时间和事件时间,对应的 TableSource 需要同时实现这两个接口:

public interface DefinedProctimeAttribute {
	@Nullable
	String getProctimeAttribute();
}

public interface DefinedRowtimeAttributes {
	List<RowtimeAttributeDescriptor> getRowtimeAttributeDescriptors();
}

其中 RowtimeAttributeDescriptor 是对事件时间的描述,包括如何提取事件时间,以及 watermark 的生成策略等:

public final class RowtimeAttributeDescriptor {
	private final String attributeName; //时间属性名称
	private final TimestampExtractor timestampExtractor; //如何提取事件时间
	private final WatermarkStrategy watermarkStrategy; //如何生成 watermark
}

尽管返回值是 List<RowtimeAttributeDescriptor>,但目前 Flink SQL 只支持单个事件时间属性。

SQL 引擎中时间属性的转换

更新 TableSchema

在将 DataStream 转换成一个 Table 的过程中,首先需要生成表结构。Table 的底层对应的是一个 QueryOperation,在这里就是 ScalaDataStreamQueryOperation (或者 JavaDataStreamQueryOperation,对应 Java API)。QueryOperation 提供了 TableSchema 和字段映射关系:

public class ScalaDataStreamQueryOperation<E> implements QueryOperation {
	private final DataStream<E> dataStream;
	private final int[] fieldIndices; //字段索引映射关系
	private final TableSchema tableSchema; //表结构
}

获得 TableSchema 的逻辑主要被封装在 FieldInfoUtils.getFieldsInfo 方法中,主要是通过解析 Expression 获得表结构中每一列对应的字段在 DataStream 中元素的索引,并得到对应字段的类型:

/**
 * Utility methods for extracting names and indices of fields from different {@link TypeInformation}s.
 */
public class FieldInfoUtils {
	private static class ExprToFieldInfo extends ApiExpressionDefaultVisitor<FieldInfo> {
		@Override
		public FieldInfo visit(UnresolvedReferenceExpression unresolvedReference) {
			return createFieldInfo(unresolvedReference, null);
		}

		@Override
		public FieldInfo visit(UnresolvedCallExpression unresolvedCall) {
			if (unresolvedCall.getFunctionDefinition() == BuiltInFunctionDefinitions.AS) {
				return visitAlias(unresolvedCall);
			} else if (isRowTimeExpression(unresolvedCall)) {
				return createRowtimeFieldInfo(unresolvedCall, null);
			} else if (isProcTimeExpression(unresolvedCall)) {
				return createProctimeFieldInfo(unresolvedCall, null);
			}

			return defaultMethod(unresolvedCall);
		}
	}

	private static boolean isRowTimeExpression(Expression origExpr) {
		return origExpr instanceof UnresolvedCallExpression &&
			((UnresolvedCallExpression) origExpr).getFunctionDefinition() == BuiltInFunctionDefinitions.ROWTIME;
	}

	private static boolean isProcTimeExpression(Expression origExpr) {
		return origExpr instanceof UnresolvedCallExpression &&
			((UnresolvedCallExpression) origExpr).getFunctionDefinition() == BuiltInFunctionDefinitions.PROCTIME;
	}

	private static FieldInfo createTimeAttributeField(
			UnresolvedReferenceExpression reference,
			TimestampKind kind, //这里的Kind是TimestampKind.PROCTIME或TimestampKind.ROWTIME
			@Nullable String alias) {
		final int idx;
		//对于时间属性,没有对应的索引,用特殊的标识
		if (kind == TimestampKind.PROCTIME) {
			idx = TimeIndicatorTypeInfo.PROCTIME_STREAM_MARKER;
		} else {
			idx = TimeIndicatorTypeInfo.ROWTIME_STREAM_MARKER;
		}

		String originalName = reference.getName();
		return new FieldInfo(
			alias != null ? alias : originalName,
			idx,
			createTimeIndicatorType(kind));
	}
}

由于时间属性对应的表达式是内置函数的调用,因而可以通过判断对应的函数定义识别出来。

时间属性并不对应 DataStream 中元素真实的字段,因此会用特殊的标识来作为索引:

public class TimeIndicatorTypeInfo extends SqlTimeTypeInfo<Timestamp> {
	public static final int ROWTIME_STREAM_MARKER = -1;
	public static final int PROCTIME_STREAM_MARKER = -2;
}

如果通过 TableSource 注册一张表,首先会通过 TableSourceValidation.validateTableSource() 验证表结构、时间属性等信息,然后会被封装为 ConnectorCatalogTable,在这里会完成 TableSchema 的更新:

public class ConnectorCatalogTable<T1, T2> extends AbstractCatalogTable {
	public static <T1> ConnectorCatalogTable source(TableSource<T1> source, boolean isBatch) {
		//更新 TableSchema
		final TableSchema tableSchema = calculateSourceSchema(source, isBatch);
		return new ConnectorCatalogTable<>(source, null, tableSchema, isBatch);
	}

	private static <T1> TableSchema calculateSourceSchema(TableSource<T1> source, boolean isBatch) {
		TableSchema tableSchema = source.getTableSchema();
		if (isBatch) {
			return tableSchema;
		}

		DataType[] types = Arrays.copyOf(tableSchema.getFieldDataTypes(), tableSchema.getFieldCount());
		String[] fieldNames = tableSchema.getFieldNames();
		//检查是否实现了 DefinedRowtimeAttributes 接口
		if (source instanceof DefinedRowtimeAttributes) {
			updateRowtimeIndicators((DefinedRowtimeAttributes) source, fieldNames, types);
		}
		//检查是否实现了 DefinedProctimeAttribute 接口
		if (source instanceof DefinedProctimeAttribute) {
			updateProctimeIndicator((DefinedProctimeAttribute) source, fieldNames, types);
		}
		return TableSchema.builder().fields(fieldNames, types).build();
	}

	private static void updateRowtimeIndicators(
			DefinedRowtimeAttributes source,
			String[] fieldNames,
			DataType[] types) {
		List<String> rowtimeAttributes = source.getRowtimeAttributeDescriptors()
			.stream()
			.map(RowtimeAttributeDescriptor::getAttributeName)
			.collect(Collectors.toList());

		for (int i = 0; i < fieldNames.length; i++) {
			if (rowtimeAttributes.contains(fieldNames[i])) {
				// bridged to timestamp for compatible flink-planner
				types[i] = new AtomicDataType(new TimestampType(true, TimestampKind.ROWTIME, 3))
						.bridgedTo(java.sql.Timestamp.class);
			}
		}
	}
}

经过更新后的 TableSchema,时间属性列的 LogicalType 就是用特殊 TimestampKind 表征的 TimestampType。

转换到 Calcite

在 DatabaseCalciteSchema 中, Flink SQL 中注册的表被转换成 Calcite 中使用的表:

class DatabaseCalciteSchema extends FlinkSchema {
	@Override
	public Table getTable(String tableName) {
		ObjectPath tablePath = new ObjectPath(databaseName, tableName);

		try {
			if (!catalog.tableExists(tablePath)) {
				return null;
			}

			CatalogBaseTable table = catalog.getTable(tablePath);

			//将 Flink Catalog 中注册的表转换为 Calcite 中的 Table
			if (table instanceof QueryOperationCatalogView) {
				QueryOperationCatalogView view = (QueryOperationCatalogView) table;
				QueryOperation operation = view.getQueryOperation();
				if (operation instanceof DataStreamQueryOperation) {
					List<String> qualifiedName = Arrays.asList(catalogName, databaseName, tableName);
					((DataStreamQueryOperation) operation).setQualifiedName(qualifiedName);
				} else if (operation instanceof RichTableSourceQueryOperation) {
					List<String> qualifiedName = Arrays.asList(catalogName, databaseName, tableName);
					((RichTableSourceQueryOperation) operation).setQualifiedName(qualifiedName);
				}
				return QueryOperationCatalogViewTable.createCalciteTable(view);
			} else if (table instanceof ConnectorCatalogTable) {
				return convertConnectorTable((ConnectorCatalogTable<?, ?>) table, tablePath);
			} else if (table instanceof CatalogTable) {
				return convertCatalogTable(tablePath, (CatalogTable) table);
			} else {
				throw new TableException("Unsupported table type: " + table);
			}
		} catch (TableNotExistException | CatalogException e) {
			// TableNotExistException should never happen, because we are checking it exists
			// via catalog.tableExists
			throw new TableException(format(
					"A failure occurred when accessing table. Table path [%s, %s, %s]",
					catalogName,
					databaseName,
					tableName), e);
		}
	}
}

在 Flink SQL 中定义的表结构 TableSchema 也会经过 FlinkTypeFactory 转换,LogicalType也会被转换成 Calcite 内部使用的 RelDataType:

/**
  * Flink specific type factory that represents the interface between Flink's [[LogicalType]]
  * and Calcite's [[RelDataType]].
  */
class FlinkTypeFactory(typeSystem: RelDataTypeSystem) extends JavaTypeFactoryImpl(typeSystem) {
  def createFieldTypeFromLogicalType(t: LogicalType): RelDataType = {
    ......
    val relType = t.getTypeRoot match {
      case LogicalTypeRoot.TIMESTAMP_WITHOUT_TIME_ZONE =>
        val timestampType = t.asInstanceOf[TimestampType]
        timestampType.getKind match {
          //时间戳类型识别出常规的时间戳和时间属性
          case TimestampKind.PROCTIME => createProctimeIndicatorType(true)
          case TimestampKind.ROWTIME => createRowtimeIndicatorType(true)
          case TimestampKind.REGULAR => createSqlType(TIMESTAMP)
        }
      case _ =>
        seenTypes.get(t) match {
          case Some(retType: RelDataType) => retType
          case None =>
            val refType = newRelDataType()
            seenTypes.put(t, refType)
            refType
        }
    }
    .....
  }
}

经过这一步转换,时间属性被转换成 TimeIndicatorRelDataType 类型。

查询计划转换

在 Calcite 优化查询计划是,会识别特殊语句中的时间属性,并转换成对应的 RelNode。例如,在 FlinkLogicalJoin 中,如果关联条件中包含了时间窗口,就会被转换为 StreamExecWindowJoin;而不包含时间窗口的 FlinkLogicalJoin 则会被转换为 StreamExecJoin。其区别就在于对时间属性的识别:

/**
  * Rule that converts non-SEMI/ANTI [[FlinkLogicalJoin]] with window bounds in join condition
  * to [[StreamExecWindowJoin]].
  */
class StreamExecWindowJoinRule
  extends ConverterRule(
    classOf[FlinkLogicalJoin],
    FlinkConventions.LOGICAL,
    FlinkConventions.STREAM_PHYSICAL,
    "StreamExecWindowJoinRule") {

  override def matches(call: RelOptRuleCall): Boolean = {

    ......

    val tableConfig = FlinkRelOptUtil.getTableConfigFromContext(join)
    //提取时间窗口的边界
    val (windowBounds, _) = WindowJoinUtil.extractWindowBoundsFromPredicate(
      join.getCondition,
      join.getLeft.getRowType.getFieldCount,
      joinRowType,
      join.getCluster.getRexBuilder,
      tableConfig)

    if (windowBounds.isDefined) {
      //如果识别到时间窗口,该规则匹配
      if (windowBounds.get.isEventTime) {
        true
      } else {
        // Check that no event-time attributes are in the input because the processing time window
        // join does not correctly hold back watermarks.
        // We rely on projection pushdown to remove unused attributes before the join.
        !joinRowType.getFieldList.exists(f => FlinkTypeFactory.isRowtimeIndicatorType(f.getType))
      }
    } else {
      // the given join does not have valid window bounds. We cannot translate it.
      false
    }
  }
}

窗口聚合也是同理,只是规则匹配中是对 HOP, TUMBLE 等窗口函数的识别。

物化时间属性

时间属性是一个逻辑上的列,因为它并不真实对应底层 DataStream 元素中具体的字段。因此,如果在计算操作中要用到时间属性列的值,就需要“物化”(materialized)时间属性。这一部分的逻辑主要是在 RelTimeIndicatorConverter 中。

对于 PROCTIME 时间属性,就是将其转换为对内置函数 FlinkSqlOperatorTable.PROCTIME_MATERIALIZE 的调用;而对于 ROWTIME 时间属性,则是对其进行一次强制的类型转换,转换为常规的时间戳。

/**
  * Helper class for shared logic of materializing time attributes in [[RelNode]] and [[RexNode]].
  */
class RexTimeIndicatorMaterializerUtils(rexBuilder: RexBuilder) {
  def materialize(expr: RexNode): RexNode = {
    if (isTimeIndicatorType(expr.getType)) {
      if (isRowtimeIndicatorType(expr.getType)) {
        // cast rowtime indicator to regular timestamp
        rexBuilder.makeAbstractCast(timestamp(expr.getType.isNullable), expr)
      } else {
        // generate proctime access
        rexBuilder.makeCall(FlinkSqlOperatorTable.PROCTIME_MATERIALIZE, expr)
      }
    } else {
      expr
    }
  }
}

生成物理执行计划

SQL 的查询计划最终需要被转换为 Flink 的算子,即生成物理执行计划。对于一个查询计划来说,首先需要进行转换的就是 Scan 操作,在里将底层的 POJO、Tuple 等对象映射为 Table 中使用的 Row。在这里就需要考虑到时间属性的映射。

首先,我们来看下在 StreamExecDataStreamScan 中是如何完成字段映射的转换的:


class StreamExecDataStreamScan(
    cluster: RelOptCluster,
    traitSet: RelTraitSet,
    table: RelOptTable,
    outputRowType: RelDataType)
  extends TableScan(cluster, traitSet, table) {
  override protected def translateToPlanInternal(
      planner: StreamPlanner): Transformation[BaseRow] = {
    val config = planner.getTableConfig
    val inputDataStream: DataStream[Any] = dataStreamTable.dataStream
    val transform = inputDataStream.getTransformation

    val rowtimeExpr = getRowtimeExpression(planner.getRelBuilder)

    // when there is row time extraction expression, we need internal conversion
    // when the physical type of the input date stream is not BaseRow, we need internal conversion.
    if (rowtimeExpr.isDefined || ScanUtil.needsConversion(dataStreamTable.dataType)) {

      // extract time if the index is -1&nbs***bsp;-2.
      val (extractElement, resetElement) =
        if (ScanUtil.hasTimeAttributeField(dataStreamTable.fieldIndexes)) {
          (s"ctx.$ELEMENT = $ELEMENT;", s"ctx.$ELEMENT = null;")
        } else {
          ("", "")
        }
      val ctx = CodeGeneratorContext(config).setOperatorBaseClass(
        classOf[AbstractProcessStreamOperator[BaseRow]])
      //生成字段映射的代码
      ScanUtil.convertToInternalRow(
        ctx,
        transform,
        dataStreamTable.fieldIndexes, //字段索引
        dataStreamTable.dataType,
        getRowType,
        getTable.getQualifiedName,
        config,
        rowtimeExpr, //RowTime 表达式
        beforeConvert = extractElement,
        afterConvert = resetElement)
    } else {
      transform.asInstanceOf[Transformation[BaseRow]]
    }
  }

  //获取 RowTime 时间属性的表达式
  private def getRowtimeExpression(relBuilder: FlinkRelBuilder): Option[RexNode] = {
    val fieldIdxs = dataStreamTable.fieldIndexes

    if (!fieldIdxs.contains(ROWTIME_STREAM_MARKER)) {
      None
    } else { 
      //根据字段映射中的特殊marker来查找 rowtime 字段
      val rowtimeField = dataStreamTable.fieldNames(fieldIdxs.indexOf(ROWTIME_STREAM_MARKER))

      // get expression to extract timestamp
      fromDataTypeToLogicalType(dataStreamTable.dataType) match {
        //如果这个datastream本身已经完成了 rowtime 时间属性的提取
        case dataType: RowType
          if dataType.getFieldNames.contains(rowtimeField) &&
              TypeCheckUtils.isRowTime(dataType.getTypeAt(dataType.getFieldIndex(rowtimeField))) =>
          // if rowtimeField already existed in the data stream, use the default rowtime
          None
        case _ =>
          // 用内置函数 StreamRecordTimestampSqlFunction 来提取 rowtime 时间属性
          // extract timestamp from StreamRecord
          Some(
            relBuilder.cast(
              relBuilder.call(new StreamRecordTimestampSqlFunction),
              relBuilder.getTypeFactory.createFieldTypeFromLogicalType(
                new TimestampType(true, TimestampKind.ROWTIME, 3)).getSqlTypeName))
      }
    }
  }
}


在进行字段映射的时候,一个关键的处理就是时间属性要怎么映射。ROWTIME 时间属性被转换为 StreamRecordTimestampSqlFunction 的调用。在前面物化时间属性的阶段,我们已经看到,PROCTIME 时间属性已经被转换为内置函数 FlinkSqlOperatorTable.PROCTIME_MATERIALIZE 的调用,而对于 ROWTIME 时间属性,只是进行了一次强制的类型转换。这主要是因为,PROCTIME 在流处理中,在不同的算子中,每一次调用都应该获取当前的系统时间;而对于 ROWTIME 而言,它的取值是固定的,因此只需要在最开始完成一次转换即可。RPOCTIME 和 ROWTIME 的取值最终都被转换成对内置函数的调用。

这个函数调用的代码生成就比较简单了,RPOCTIME 转换成 context.timerService().currentProcessingTime(),ROWTIME 转换成 context.timestamp(),具体的代码生成可以参考 GenerateUtils。

由于 PROCTIME 并不需要在 Scan 节点进行物化,因此在这里直接用 null 值替代,在后续需要的时候重新进行计算。

而对于从 TableSource 注册而来的表,由于它不像 DataStream 那样,在定义 ROWTIME 时间属性之前已经完成了 timestamp 和 watermark 的指定。因此在转换 TableSource 的过程中,如果定义了 ROWTIME 时间属性,除了需要提取 ROWTIME 时间属性的值以外,还需要指定 watermark。

RowtimeAttributeDescriptor 是对 ROWTIME 时间属性的描述,包括如何提取事件时间的 TimestampExtractor:

/**
 * The {@link FieldComputer} interface returns an expression to compute the field of the table
 * schema of a {@link TableSource} from one&nbs***bsp;more fields of the {@link TableSource}'s return type.
 *
 * @param <T> The result type of the provided expression.
 */
public interface FieldComputer<T> {
	String[] getArgumentFields();


	TypeInformation<T> getReturnType();

	void validateArgumentFields(TypeInformation<?>[] argumentFieldTypes);

	/**
	 * Returns the {@link Expression} that computes the value of the field.
	 *
	 * @param fieldAccesses Field access expressions for the argument fields.
	 * @return The expression to extract the timestamp from the {@link TableSource} return type.
	 */
	Expression getExpression(ResolvedFieldReference[] fieldAccesses);
}

//Provides an expression to extract the timestamp for a rowtime attribute.
public abstract class TimestampExtractor implements FieldComputer<Long>, Serializable, Descriptor {
	@Override
	public TypeInformation<Long> getReturnType() {
		return Types.LONG; //返回类型是 Long
	}
}

通常提取 ROWTIME 的方式是根据已有的字段来生成,即:
public final class ExistingField extends TimestampExtractor {
	private String field;
	
	/**
	 * Returns an {@link Expression} that casts a {@link Long}, {@link Timestamp},&nbs***bsp;* timestamp formatted {@link String} field (e.g., "2018-05-28 12:34:56.000")
	 * into a rowtime attribute.
	 */
	@Override
	public Expression getExpression(ResolvedFieldReference[] fieldAccesses) {
		ResolvedFieldReference fieldAccess = fieldAccesses[0];
		DataType type = fromLegacyInfoToDataType(fieldAccess.resultType());

		//字段引用的表达式
		FieldReferenceExpression fieldReferenceExpr = new FieldReferenceExpression(
				fieldAccess.name(),
				type,
				0,
				fieldAccess.fieldIndex());
		//支持的输入字段类型,包括 BIGINT、TIMESTAMP_WITHOUT_TIME_ZONE 和 VARCHAR
		switch (type.getLogicalType().getTypeRoot()) {
			case BIGINT:
			case TIMESTAMP_WITHOUT_TIME_ZONE:
				//直接引用相应的字段即可
				return fieldReferenceExpr;
			case VARCHAR:
				//进行一次类型转换
				DataType outputType = TIMESTAMP(3).bridgedTo(Timestamp.class);
				return new CallExpression(
						CAST,
						Arrays.asList(fieldReferenceExpr, typeLiteral(outputType)),
						outputType);
			default:
				throw new RuntimeException("Unsupport type: " + type);
		}
	}
}

TimestampExtractor 提供的提取时间属性的 Expression, 要先被转换为 RexNode 之后才可以被用在代码生成中,通过 TableSourceUtil.getRowtimeExtractionExpression 完成。如同前面提到对 StreamRecordTimestampSqlFunction 调用,这两种获得时间属性的的方式本质上是一样的,只是调用的函数不一致。

在完成了字段映射和时间属性的提取后,还需要指定 watermark,这正是 RowtimeAttributeDescriptor 中 WatermarkStrategy 的用处:

class StreamExecTableSourceScan(
    cluster: RelOptCluster,
    traitSet: RelTraitSet,
    relOptTable: FlinkRelOptTable)
  extends PhysicalTableSourceScan(cluster, traitSet, relOptTable) {

  override protected def translateToPlanInternal(
      planner: StreamPlanner): Transformation[BaseRow] = {
    val config = planner.getTableConfig
    val inputTransform = getSourceTransformation(planner.getExecEnv)

    //1. 字段映射和时间提取
    ......

    //2. 指定watermark
    val withWatermarks = if (rowtimeDesc.isDefined) {
      val rowtimeFieldIdx = getRowType.getFieldNames.indexOf(rowtimeDesc.get.getAttributeName)
      val watermarkStrategy = rowtimeDesc.get.getWatermarkStrategy
      watermarkStrategy match {
        case p: PeriodicWatermarkAssigner =>
          val watermarkGenerator = new PeriodicWatermarkAssignerWrapper(rowtimeFieldIdx, p)
          ingestedTable.assignTimestampsAndWatermarks(watermarkGenerator)
        case p: PunctuatedWatermarkAssigner =>
          val watermarkGenerator =
            new PunctuatedWatermarkAssignerWrapper(rowtimeFieldIdx, p, producedDataType)
          ingestedTable.assignTimestampsAndWatermarks(watermarkGenerator)
        case _: PreserveWatermarks =>
          // The watermarks have already been provided by the underlying DataStream.
          ingestedTable
      }
    } else {
      // No need to generate watermarks if no rowtime attribute is specified.
      ingestedTable
    }
    withWatermarks.getTransformation

  }
}

到这里,就完成了 ROWTIME 时间属性的提取的 watermark 的生成。 PROCTIME 在使用时按需要进行物化。

小结

时间属性是广泛应用在窗口操作中,是流式 SQL 处理中非常重要的概念。本文对 Flink SQL 中时间属性的使用方法和具体实现进行了介绍。

参考



转发自:https://blog.jrwang.me/2019/2019-09-16-flink-sourcecode-sql-time-attribute.md/