Versions Compared

Old Version 32

changes.mady.by.user Xuannan Su

Saved on

compared with

New Version 33

changes.mady.by.user Xuannan Su

Saved on

Key

  • This line was added.
  • This line was removed.
  • Formatting was changed.

...

Code Block
languagejava
titlecache() and invalidateCache() API
  /**
    * Cache this table to builtin table service or the specified customized table service.
    *
    * This method provides a hint to Flink that the current table maybe reused later so a
    * cache should be created to avoid regenerating this table.
    *
    * The following code snippet gives an example of how this method could be used.
    *
    * {{{
    *   valTable t = tEnv.fromCollection(data).as('country, 'color, 'count)
    *
    *   valCachedTable t1 = t.filter('count < 100).cache()
    *   // t1 is cached after it is computed for the first time.
    *   t1.execute().print()
    *
    *   // When t1 is used again to compute t2, it may not be re-computed.
    *   valTable t2 = t1.groupBy('country).select('country, 'count.sum as 'sum)
    *   t2.execute().print()
    *
    *   // Similarly when t1 is used again to compute t3, it may not be re-computed.
    *   valTable t3 = t1.groupBy('color).select('color, 'count.avg as 'avg)
    *   t3.execute().print()
    *
    * }}}
    *
    * @note Flink optimizer may decide to not use the cache if doing that will accelerate the
    * processing, or if the cache is no longer available for reasons such as the cache has
    * been invalidated.
    * @note The table cache could be create lazily. That means the cache may be created at 
    * the first time when the cached table is computed.
    * @note The table cache will be cleared when the user program exits.
	* @note This method is only supported in batch table and it is treated as No-Op for stream table
    *
    * @return the current table with a cache hint. The original table reference is not modified
    *               by the execution of this method. If this method is called on a table with cache 
	*               hint, the same table object will return.
    */
   defTable cache(): Table

;

2. Add a CachedTable interface extend Table

Code Block
languagejava
titleCachedTable
public interface CachedTable extends Table {
  /**
    * Manually invalidate the cache of this table to release the physical resources. Users are
    * not required to invoke this method to release physical resource unless they want to. The
    * table caches are cleared when user program exits.
    *
    * @note After invalidated, the cache may be re-created if this table is used again.
    */
  defvoid invalidateCache(): Unit;
}

32. Add a close method to the TableEnvironment

...

As mentioned in the motivation section. The key idea of the FLIP is to allow the intermediate process results to be cached, so later references to that result does do not result in require duplicate computation. To achieve that, we need to introduce Cached Tables.

The cached tables are tables whose contents are saved by Flink as the user application runs. A cached Table can be created in two waysthe following way:

  • Users can call cache() method on a table to explicitly tell Flink to cache a Table.
  • The cache() method returns a new Table CachedTable object with a flag set.
  • The cache() method does not execute eagerly. Instead, the table will be cached when the DAG that contains the cached table runs.
Code Block
languagejava
TableEnvironment tEnv = ...
Table t1 = ...
TableCachedTable t2 = t1.cache()
...
tEnvt2.execute().print() // t1 is cached.
Table t3 = t1.select(...) // cache will NOT be used.
Table t4 = t2.select(...) // cache will be used.
...
// The following two lines of code are equivalent
t1.invalidateCache() // cache will be released
t2.invalidateCache() // cache will be released
...
t1.execute().print() // cache will NOT be recreated
t2.execute().print() // cache will be recreated

...

When TableEnvironment is closed, the resources consumed by the cached tables will also be released. This usually happens when user application exits.

...

Sometimes users may want to release the resource used by a cached table before the application exits. In this case, users can call invalidateCache() on a table. This will immediately release the resources used to cache that table.

In some rare cases, users may want to explicitly ignore the cached intermediate result. In this case, users need to give an explicit hint, such as:

table.hint("ignoreCache")

Right now Flink does not have a hint mechanism yet. So before such a hint mechanism is available. Users are not able to explicitly ignore a cached intermediate result.

Theoretically speaking, users can also cache a streaming table. The semantic will be storing the result somewhere (potentially with a TTL). However, caching a streaming table is usually not that useful. For simplicity, we would not support stream table caching in the first implementation. When cache() is invoked on a stream table, it will be treated as a No-Op. This leaves us room to add caching for stream tables in the future without asking users to change their code.

Implementation Details

To let the feature available out of the box, a default file system based cache service will be provided, which utilizes the cluster partition implemented in FLIP-67. This section describes the implementation details of the default table service.

Although the implementation details are transparent to the users, there are some related changes to make the default implementation work.

The architecture is illustrated below:

Image Removed

Each cached table consists of two pieces of information:

  • Table metadata - name, location, etc.
  • Table contents - the actual contents of the table

The default table service stores the metadata in the client (e.g. TableEnvironment) and the actual contents are stored in the Task Managers as cluster partitions(FLIP-67).

The end to end process is the following:

Step 1: Execute JOB_1 (write cached tables)

  1. Users call table.cache(), the client
    1. adds a Sink to the cached node in the DAG. By adding the Sink to the cached node, we can make sure that the optimizer would not affect the cached node. The default IntermediateResultStorage creates a CacheSink.
    2. generate an IntermediateResultId
    3. passes the IntermediateResultId created all the way from RelNode down to the Operators in the JobVertex
    4. Set IntermediateDataSetId to IntermediateResultId
  2. The JobGraphGenerator recognizes the CacheSink, removes the Sink and sets the result partition type of the producer to BLOCKING_PERSISTENT
  3. The client submits the job
  4. JobMaster executes the job like usual. After the job finishes, the TaskExecutor promotes the BLOCKING_PERSISTENT result partitions to cluster partitions instead of releasing them (Implemented in FLIP-67)
  5. After the job finishes, JobMaster reports the information of the cluster partition (ClusterPartitionDescriptor) back to the client in form of a mapping of [IntermediateDataSetId -> [ClusterPartitionDescriptor]]
    1. The ClusterPartitionDescriptor should include a ShuffleDescriptor and some metadata, i.e. numberOfSubpartitions and partitionType
    2. The ClusterPartitionDescriptor will be serialized before sending back to the client and only be deserialized in the JobGraphGenerator
    3. The table environment on client maintain the mapping of CachedTable -> (IntermediateResultId, [ClusterPartitionDescriptor])

Step 2: Execute JOB_2 (read cached tables)

  1. Later on, when the client submits another job whose DAG contains a cached node, the client
    1. looks up the available intermediate results
    2. creates a Source node(CacheSource) that contains the ClusterPartitionDescriptor
    3. replace the subtree of the cached node with the source node created
  2. The JobGraphGenerator sees a CacheSource node, sets its downstream node’s cluster partition input and sets the operator/driver to NoOp. It will then remove itself from the JobGraph
    1. The parallelism is set to the maximum number of subpartitions among the cluster partitions. It ensures that all the subpartitions are read by the NoOp vertex
    2. The shipStrategyName field in the output edge of the NoOp vertex contains information about how the record should be partitioned
  3. The clients submit the job
  4. JobMaster does the following if the JobVertex has cluster partition input set 
    1. It assumes Scheduler understands the cluster partition location
    2. Create InputGateDeploymentDescriptor with the ShuffleMaster
    3. assign the result partitions to each subtask based on locality
  5. Task managers will run the given tasks as usual

Clean up 

  1. When the application exits, all the Task Managers will exit and the intermediate results will be released.
  1. Users invoke Table.invalidateCache()
    1. Clients remove the intermediate result entry in the TableEnvironment.
  2. Clients delete the corresponding cluster partitions with the REST API provided in FLIP-67.
  3. The cluster partitions will then be released by the Task Managers that hold the cluster partitions.

Implementation Details

To let the feature available out of the box, a default file system based cache service will be provided, which utilizes the cluster partition implemented in FLIP-67. This section describes the implementation details of the default table service.

Although the implementation details are transparent to the users, there are some related changes to make the default implementation work.

The architecture is illustrated below:


Each cached table consists of two pieces of information:

  • Table metadata - name, location, etc.
  • Table contents - the actual contents of the table

The default table service stores the metadata in the client (e.g. TableEnvironment) and the actual contents are stored in the Task Managers as cluster partitions(FLIP-67).

The end to end process is the following:

Image Added

Step 1: Execute JOB_1 (write cached tables)

  1. Users call table.cache(), the client
    1. Wrap the operation of the table with CacheOperation. The CacheOpeartion contains a generated IntermediateResultId
    2. While translating the operations to rel nodes, the planner adds a CacheSink, which contains the IntermediateResultId, to the cached node in the DAG. By adding a sink to the cached node, we can make sure that the optimizer would not affect the cached node
    3. The CacheSink is treated as a normal sink when the planner optimizes and translates the rel nodes to transformation
  2. When the Executor(StreamGraphGenerator) translates the transformation to stream graph, it recognizes the CacheSink, removes the CacheSink, and sets the cache flag of the upstream node.
  3. If the JobGraphGenerator see the node with the cache flag set, it sets the result partition type to BLOCKING_PERSISTENT
  4. The client submits the job
  5. JobMaster executes the job as usual. After the job finishes, the TaskExecutor promotes the BLOCKING_PERSISTENT result partitions to cluster partitions instead of releasing them (Implemented in FLIP-67)
  6. After the job finishes, JobMaster reports the information of the cluster partition (ClusterPartitionDescriptor) back to the client in form of a mapping of [IntermediateDataSetId -> [ClusterPartitionDescriptor]]
    1. The ClusterPartitionDescriptor should include a ShuffleDescriptor and some metadata, i.e. numberOfSubpartitions and partitionType
    2. The ClusterPartitionDescriptor will be serialized before sending back to the client via JobResult and only be deserialized in the JobGraphGenerator
    3. The CatalogManager in the table environment maintain the mapping of CachedTable -> (IntermediateResultId, [ClusterPartitionDescriptor])

Step 2: Execute JOB_2 (read cached tables)

  1. Later on, when the client submits another job whose DAG contains a CacheOperation, the planner
    1. looks up the available intermediate results in the CatalogManager
    2. creates a Source node(CacheSource) that contains the ClusterPartitionDescriptor
    3. replace the subtree of the cached node with the source node created
    4. The CacheSource is treated as a normal source when the planner optimizes and translates the rel nodes to transformation
  2. The StreamGraphGenerator recognizes the CacheSource and includes the ClusterPartitionDescriptor in the StreamNode
  3. When the JobGraphGenerator sees the StreamNode that contains the ClusterPartitionDescriptor, it will include the ClusterPartitionDescriptor in the JobVertex and set the operator/driver to NoOp.
    1. The parallelism is set to the maximum number of subpartitions among the cluster partitions to ensure that all the subpartitions are read by the NoOp vertex
  4. The clients submit the job
  5. JobMaster does the following if the JobVertex contains the ClusterPartitionDescriptor
    1. It assumes Scheduler understands the cluster partition location
    2. Create InputGateDeploymentDescriptor with the ShuffleMaster
    3. assign the result partitions to each subtask based on locality
  6. Task managers will run the given tasks as usual

Clean up 

  1. When the application exits, all the Task Managers will exit and the intermediate results will be released.
  1. Users invoke CachedTable.invalidateCache()
    1. Clients remove the intermediate result entry in the TableEnvironment.
  2. Clients delete the corresponding cluster partitions with the REST API provided in FLIP-67.
  3. The cluster partitions will then be released by the Task Managers that hold the cluster partitions.

Please refer to FLIP-67 for the implementation detail of steps 2 and 3.

The cached table utilizes the cluster partition implemented in FLIP-67 to store the intermediate result in the TMs. Therefore, it is possible that the TM could run out of space to hold the intermediate result if clients cache too many tables without releasing. When it happens, the job will fail with an exception. For simplicity, we would not implement an automatic mechanism to handle such failure in the first implementation. It is up to the user to decide what to do when the failure happens. In the future, we could introduce some kind of eviction policy to release some cluster partition when the failure happens and re-run the job.

In the per-job mode cluster, a cluster will be spun up on every submitted job and tore down when the submitted job finished. All lingering resources (files, etc) are cleared up, including the cluster partition in the TMs. Therefore, the cached table will not work in the Per-Job mode cluster. When a job that will create some cache tables is submitted to in Per-Job mode, the returned ClusterPartitionDescriptors will be ignored so that the planner will not attempt to replace the subtree of the cache nodePlease refer to FLIP-67 for the implementation detail of steps 2 and 3.

Repartition is needed when the cache consumer requires the input data to be partitioned in a specific way, i.e. hash partition, custom partition. When the JobGraphGenerator generates the job graph, it introduces a NoOp job vertex as the upstream vertex of the cache consumer and maintains the shipStrategyName of the output job edge. During execution, the task executor will make sure that the data is repartitioned.

...

The above process is transparent to the users.

In order to implement the default intermediate result storage, the following changes are needed.

  • New result partition type: BLOCKING_PERSISTENT
    • A BLOCKING_PERSISTENT result partition will not be deleted when the job exits.
  • JobMaster reports IntermediateDataSetID to ClusterPartitionDescriptor mapping to Client.
  • Add IntermediateDataSetID to StreamTransformation and Operator
  • TableEnvironment stores the Table → (IntermediateResultId, [ClusterPartitionDescriptor]) mapping
  • Client replaces the source of the cached node before optimization.
  • Client adds a sink to the node that should be cached.

To explain how the optimizer affects the cache node, let To explain how the optimizer affects the cache node, let look at a very simple DAG, where one scan node followed by a filter node, as shown below. The optimizer can push the filter to the scan node such that the scan node will produce fewer data to the downstream node. However, such optimization will affect to the downstream node. However, such optimization will affect the result of the scan node, which is an undesired behavior if we want to cache the scan node.

Image Added

To solve the problem above, when users explicitly cache a table, we will change the DAG implicitly by adding a special sink node to the cache table, as shown below. By doing this, we are able to ensure that the result of the scan node, which is an undesired behavior if we want to cache the scan node.

Image Removed

To solve the problem above, when users explicitly cache a table, we will change the DAG implicitly by adding a special sink node to the cache table, as shown below. By doing this, we are able to ensure that the result of the cache node will not be affected by the optimizer and we can identify the job vertex that produces the cache table in JobGraphGenerator by the special sink. With Blink planner, when a DAG has multiple sinks, the multi sink optimization will break the DAG into multiple RelNodeBlocks, the cache node will be the output node of one of those RelNodeBlocks. In our example, it will be broken into three RelNodeBlocks. Then, the optimizer will run independently on each of those RelNodeBlocks so that the optimizer will not affect the result of the cache node.

...

cache node will not be affected by the optimizer and we can identify the job vertex that produces the cache table in JobGraphGenerator by the special sink. With Blink planner, when a DAG has multiple sinks, the multi sink optimization will break the DAG into multiple RelNodeBlocks, the cache node will be the output node of one of those RelNodeBlocks. In our example, it will be broken into three RelNodeBlocks. Then, the optimizer will run independently on each of those RelNodeBlocks so that the optimizer will not affect the result of the cache node.

Image Added

In order to implement the default intermediate result storage, the following changes are needed.

  • New result partition type: BLOCKING_PERSISTENT
    • A BLOCKING_PERSISTENT result partition will not be deleted when the job exits.
  • JobMaster reports IntermediateDataSetID to ClusterPartitionDescriptor mapping to Client.
  • Add IntermediateDataSetID to StreamTransformation and Operator
  • TableEnvironment stores the Table → (IntermediateResultId, [ClusterPartitionDescriptor]) mapping
  • Client replaces the source of the cached node before optimization.
  • Client adds a sink to the node that should be cached.

Future works

In some cases, users may want to plugin their own cache service. In the future, we could add support for that.

Some API changes will be needed to support customized cache service. We will start another FLIP to discuss that. The change should not be much. Curious readers can read the google doc for some idea.

Auto cache allows the BLOCKING shuffle boundaries to be persisted for later usage. It further relieves the users from thinking about when to explicitly cache in the program.

As of now DataStream only supports stream processing. There is some idea of supporting both Stream and Batch (as finite stream) in the program.

DataStream. Once we do that, we can add the cache API to DataStream as well.

Theoretically speaking, users can also cache a streaming table. The semantic will be storing the result somewhere (potentially with a TTL). However, caching a streaming table is usually not that useful. For simplicity, we would not support stream table caching in the first implementation. When cache() is invoked on a stream table, it will be treated as a No-Op. This leaves us room to add caching for stream tables in the future without asking users to change their codeAs of now DataStream only supports stream processing. There is some idea of supporting both Stream and Batch (as finite stream) in DataStream. Once we do that, we can add the cache API to DataStream as well.

Compatibility, Deprecation, and Migration Plan

This FLIP proposes a new feature in Flink. It is fully backwards compatible.

...