//构造hive catalog
		String name = "myhive";
		String defaultDatabase = "default";
		String hiveConfDir = "/Users/user/work/hive/conf"; // a local path
		String version = "3.1.2";

		HiveCatalog hive = new HiveCatalog(name, defaultDatabase, hiveConfDir, version);
		tEnv.registerCatalog("myhive", hive);
		tEnv.useCatalog("myhive");
		tEnv.getConfig().setSqlDialect(SqlDialect.HIVE);
		tEnv.useDatabase("db1");

		tEnv.createTemporaryView("kafka_source_table", dataStream);

		String insertSql = "insert into  hive.db1.fs_table SELECT userId, amount, " +
		                   " DATE_FORMAT(ts, 'yyyy-MM-dd'), DATE_FORMAT(ts, 'HH'), DATE_FORMAT(ts, 'mm') FROM kafka_source_table";
		tEnv.executeSql(insertSql);

/**
 * Table sink to write to Hive tables.
 */
public class HiveTableSink implements AppendStreamTableSink, PartitionableTableSink, OverwritableTableSink {

	private static final Logger LOG = LoggerFactory.getLogger(HiveTableSink.class);

	private final boolean userMrWriter;
	//是否有界，用来区分是批处理还是流处理
	private final boolean isBounded;
	private final JobConf jobConf;
	private final CatalogTable catalogTable;
	private final ObjectIdentifier identifier;
	private final TableSchema tableSchema;
	private final String hiveVersion;
	private final HiveShim hiveShim;

	private LinkedHashMap<String, String> staticPartitionSpec = new LinkedHashMap<>();

	private boolean overwrite = false;
	private boolean dynamicGrouping = false;

hiveShim = HiveShimLoader.loadHiveShim(hiveVersion);

		String[] partitionColumns = getPartitionKeys().toArray(new String[0]);
		String dbName = identifier.getDatabaseName();
		String tableName = identifier.getObjectName();
		try (HiveMetastoreClientWrapper client = HiveMetastoreClientFactory.create(
				new HiveConf(jobConf, HiveConf.class), hiveVersion)) {
			Table table = client.getTable(dbName, tableName);
			StorageDescriptor sd = table.getSd();

if (isBounded){
   ......
   //batch
    
} else {
   ......
   //streaming
    
}

    //桶分配器
	TableBucketAssigner assigner = new TableBucketAssigner(partComputer);
	
	//滚动策略
	TableRollingPolicy rollingPolicy = new TableRollingPolicy(
						true,
						conf.get(SINK_ROLLING_POLICY_FILE_SIZE).getBytes(),
						conf.get(SINK_ROLLING_POLICY_ROLLOVER_INTERVAL).toMillis());

    //构造bulkFactory
	Optional<BulkWriter.Factory<RowData>> bulkFactory = createBulkWriterFactory(partitionColumns, sd);

				if (userMrWriter || !bulkFactory.isPresent()) {
					HiveBulkWriterFactory hadoopBulkFactory = new HiveBulkWriterFactory(recordWriterFactory);
					builder = new HadoopPathBasedBulkFormatBuilder<>(
							new Path(sd.getLocation()), hadoopBulkFactory, jobConf, assigner)
							.withRollingPolicy(rollingPolicy)
							.withOutputFileConfig(outputFileConfig);
					LOG.info("Hive streaming sink: Use MapReduce RecordWriter writer.");
				} else {
					builder = StreamingFileSink.forBulkFormat(
							new org.apache.flink.core.fs.Path(sd.getLocation()),
							new FileSystemTableSink.ProjectionBulkFactory(bulkFactory.get(), partComputer))
							.withBucketAssigner(assigner)
							.withRollingPolicy(rollingPolicy)
							.withOutputFileConfig(outputFileConfig);
					LOG.info("Hive streaming sink: Use native parquet&orc writer.");
				}

		StreamingFileWriter fileWriter = new StreamingFileWriter(
				rollingCheckInterval,
				bucketsBuilder);
		DataStream<CommitMessage> writerStream = inputStream.transform(
				StreamingFileWriter.class.getSimpleName(),
				TypeExtractor.createTypeInfo(CommitMessage.class),
				fileWriter).setParallelism(inputStream.getParallelism());

		DataStream<?> returnStream = writerStream;

		// save committer when we don't need it.
		if (partitionKeys.size() > 0 && conf.contains(SINK_PARTITION_COMMIT_POLICY_KIND)) {
			StreamingFileCommitter committer = new StreamingFileCommitter(
					path, tableIdentifier, partitionKeys, msFactory, fsFactory, conf);
			returnStream = writerStream
					.transform(StreamingFileCommitter.class.getSimpleName(), Types.VOID, committer)
					.setParallelism(1)
					.setMaxParallelism(1);
		}

public class StreamingFileWriter extends AbstractStreamOperator<CommitMessage>
		implements OneInputStreamOperator<RowData, CommitMessage>, BoundedOneInput{
		
	@Override
	public void initializeState(StateInitializationContext context) throws Exception {
	    .........................
	}		

		@Override
	public void snapshotState(StateSnapshotContext context) throws Exception {
	  .........................
	}

	@Override
	public void processWatermark(Watermark mark) throws Exception {
	  .........................
	}

	@Override
	public void processElement(StreamRecord<RowData> element) throws Exception {
	  .........................
	}

	/**
	 * Commit up to this checkpoint id, also send inactive partitions to downstream for committing.
	 */
	@Override
	public void notifyCheckpointComplete(long checkpointId) throws Exception {
		  .........................
	}
	    
		@Override
	public void endInput() throws Exception {
		  .........................
	}

	@Override
	public void dispose() throws Exception {
	  .........................
	}	    
		    

		   
}

	private void commitUpToCheckpoint(long checkpointId) throws Exception {
		helper.commitUpToCheckpoint(checkpointId);
		CommitMessage message = new CommitMessage(
				checkpointId,
				getRuntimeContext().getIndexOfThisSubtask(),
				getRuntimeContext().getNumberOfParallelSubtasks(),
				new ArrayList<>(inactivePartitions));
		output.collect(new StreamRecord<>(message));
		inactivePartitions.clear();
	}

	@Override
	public void initializeState(StateInitializationContext context) throws Exception {
        ..........................
		buckets.setBucketLifeCycleListener(new BucketLifeCycleListener<RowData, String>() {

			@Override
			public void bucketCreated(Bucket<RowData, String> bucket) {
			}

			@Override
			public void bucketInactive(Bucket<RowData, String> bucket) {
				inactivePartitions.add(bucket.getBucketId());
			}
		});
	}

	@Override
	public void processElement(StreamRecord<CommitMessage> element) throws Exception {
		CommitMessage message = element.getValue();
		for (String partition : message.partitions) {
			trigger.addPartition(partition);
		}

		if (taskTracker == null) {
			taskTracker = new TaskTracker(message.numberOfTasks);
		}
		boolean needCommit = taskTracker.add(message.checkpointId, message.taskId);
		if (needCommit) {
			commitPartitions(message.checkpointId);
		}
	}

	private void commitPartitions(long checkpointId) throws Exception {
		List<String> partitions = checkpointId == Long.MAX_VALUE ?
				trigger.endInput() :
				trigger.committablePartitions(checkpointId);
		if (partitions.isEmpty()) {
			return;
		}

		try (TableMetaStoreFactory.TableMetaStore metaStore = metaStoreFactory.createTableMetaStore()) {
			for (String partition : partitions) {
				LinkedHashMap<String, String> partSpec = extractPartitionSpecFromPath(new Path(partition));
				LOG.info("Partition {} of table {} is ready to be committed", partSpec, tableIdentifier);
				Path path = new Path(locationPath, generatePartitionPath(partSpec));
				PartitionCommitPolicy.Context context = new PolicyContext(
						new ArrayList<>(partSpec.values()), path);
				for (PartitionCommitPolicy policy : policies) {
					if (policy instanceof MetastoreCommitPolicy) {
						((MetastoreCommitPolicy) policy).setMetastore(metaStore);
					}
					policy.commit(context);
				}
			}
		}
	}