import org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator; //导入依赖的package包/类
private void generateNodeLocalHash(StreamNode node, Hasher hasher, int id) {
	hasher.putInt(id);

	hasher.putInt(node.getParallelism());

	if (node.getOperator() instanceof AbstractUdfStreamOperator) {
		String udfClassName = ((AbstractUdfStreamOperator<?, ?>) node.getOperator())
				.getUserFunction().getClass().getName();

		hasher.putString(udfClassName, Charset.forName("UTF-8"));
	}
}
 

import org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator; //导入依赖的package包/类
@SuppressWarnings("unchecked")
private static <T> SourceFunction<T> getFunctionFromDataSource(DataStreamSource<T> dataStreamSource) {
	dataStreamSource.addSink(new DiscardingSink<T>());
	AbstractUdfStreamOperator<?, ?> operator =
			(AbstractUdfStreamOperator<?, ?>) getOperatorFromDataStream(dataStreamSource);
	return (SourceFunction<T>) operator.getUserFunction();
}
 

import org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator; //导入依赖的package包/类
/**
 * Generates a deterministic hash from node-local properties and input and
 * output edges.
 */
private byte[] generateDeterministicHash(
		StreamNode node,
		Hasher hasher,
		Map<Integer, byte[]> hashes,
		boolean isChainingEnabled) {

	// Include stream node to hash. We use the current size of the computed
	// hashes as the ID. We cannot use the node's ID, because it is
	// assigned from a static counter. This will result in two identical
	// programs having different hashes.
	generateNodeLocalHash(node, hasher, hashes.size());

	// Include chained nodes to hash
	for (StreamEdge outEdge : node.getOutEdges()) {
		if (isChainable(outEdge, isChainingEnabled)) {
			StreamNode chainedNode = outEdge.getTargetVertex();

			// Use the hash size again, because the nodes are chained to
			// this node. This does not add a hash for the chained nodes.
			generateNodeLocalHash(chainedNode, hasher, hashes.size());
		}
	}

	byte[] hash = hasher.hash().asBytes();

	// Make sure that all input nodes have their hash set before entering
	// this loop (calling this method).
	for (StreamEdge inEdge : node.getInEdges()) {
		byte[] otherHash = hashes.get(inEdge.getSourceId());

		// Sanity check
		if (otherHash == null) {
			throw new IllegalStateException("Missing hash for input node "
					+ inEdge.getSourceVertex() + ". Cannot generate hash for "
					+ node + ".");
		}

		for (int j = 0; j < hash.length; j++) {
			hash[j] = (byte) (hash[j] * 37 ^ otherHash[j]);
		}
	}

	if (LOG.isDebugEnabled()) {
		String udfClassName = "";
		if (node.getOperator() instanceof AbstractUdfStreamOperator) {
			udfClassName = ((AbstractUdfStreamOperator<?, ?>) node.getOperator())
					.getUserFunction().getClass().getName();
		}

		LOG.debug("Generated hash '" + byteToHexString(hash) + "' for node " +
				"'" + node.toString() + "' {id: " + node.getId() + ", " +
				"parallelism: " + node.getParallelism() + ", " +
				"user function: " + udfClassName + "}");
	}

	return hash;
}
 

import org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator; //导入依赖的package包/类
private static Function getFunctionForDataStream(DataStream<?> dataStream) {
	AbstractUdfStreamOperator<?, ?> operator =
			(AbstractUdfStreamOperator<?, ?>) getOperatorForDataStream(dataStream);
	return operator.getUserFunction();
}