parameters: (context)
docstring: The entry method called by the Deep Learning on Flink framework
      :param context: The context passe by the framework, which could be used to
      construct the TFContext
docstring_summary: The entry method called by the Deep Learning on Flink framework
      :param context: The context passe by the framework, which could be used to
      construct the TFContext
docstring_tokens: The entry method called by the Deep Learning on Flink framework : param context : The context passe by the framework which could be used to construct the TFContext
function: def stream_train(context):
      """
      The entry method called by the Deep Learning on Flink framework
      :param context: The context passe by the framework, which could be used to
      construct the TFContext
      """
      from flink_ml_tensorflow.tensorflow_context import TFContext
  
      # Set the TF_CONFIG for distributed training
      tf_context = TFContext(context)
      cluster = tf_context.to_tf_cluster(tf_context.properties["cluster"])
      os.environ['TF_CONFIG'] = json.dumps({
          'cluster': cluster,
          'task': {'type': tf_context.get_role_name(),
                   'index': tf_context.get_index()}
      })
      print(os.environ['TF_CONFIG'])
  
      model_save_path = tf_context.get_property("model_save_path")
  
      def stream_dataset() -> tf.data.Dataset:
          """
          Return a DataSet that read from Flink for model training
          """
          def parse_csv(value):
              x, y = tf.io.decode_csv(value, record_defaults=[[0.], [0.]])
              return x, y
  
          dataset = tf_context.flink_stream_dataset() \
              .map(parse_csv).repeat(1) \
              .batch(32)
          if is_tf2:
              option = tf.data.Options()
              option.experimental_distribute.auto_shard_policy = \
                  tf.data.experimental.AutoShardPolicy.OFF
              dataset = dataset.with_options(option)
          return dataset
  
      train(stream_dataset, model_save_path)
function_tokens: def stream_train ( context ) : from flink_ml_tensorflow . tensorflow_context import TFContext # Set the TF_CONFIG for distributed training tf_context = TFContext ( context ) cluster = tf_context . to_tf_cluster ( tf_context . properties [ "cluster" ] ) os . environ [ 'TF_CONFIG' ] = json . dumps ( { 'cluster' : cluster , 'task' : { 'type' : tf_context . get_role_name ( ) , 'index' : tf_context . get_index ( ) } } ) print ( os . environ [ 'TF_CONFIG' ] ) model_save_path = tf_context . get_property ( "model_save_path" ) def stream_dataset ( ) -> tf . data . Dataset : """
          Return a DataSet that read from Flink for model training
          """ def parse_csv ( value ) : x , y = tf . io . decode_csv ( value , record_defaults = [ [ 0. ] , [ 0. ] ] ) return x , y dataset = tf_context . flink_stream_dataset ( ) . map ( parse_csv ) . repeat ( 1 ) . batch ( 32 ) if is_tf2 : option = tf . data . Options ( ) option . experimental_distribute . auto_shard_policy = tf . data . experimental . AutoShardPolicy . OFF dataset = dataset . with_options ( option ) return dataset train ( stream_dataset , model_save_path )