flink-issues mailing list archives

Site index · List index
Message view « Date » · « Thread »
Top « Date » · « Thread »
From "ASF GitHub Bot (JIRA)" <j...@apache.org>
Subject [jira] [Commented] (FLINK-5290) Ensure backwards compatibility of the hashes used to generate JobVertexIds
Date Mon, 12 Dec 2016 18:02:58 GMT

    [ https://issues.apache.org/jira/browse/FLINK-5290?page=com.atlassian.jira.plugin.system.issuetabpanels:comment-tabpanel&focusedCommentId=15742622#comment-15742622
] 

ASF GitHub Bot commented on FLINK-5290:
---------------------------------------

Github user aljoscha commented on a diff in the pull request:

    https://github.com/apache/flink/pull/2966#discussion_r92003177
  
    --- Diff: flink-streaming-java/src/main/java/org/apache/flink/streaming/api/graph/DefaultStreamGraphHasher.java
---
    @@ -0,0 +1,313 @@
    +/*
    + * Licensed to the Apache Software Foundation (ASF) under one
    + * or more contributor license agreements.  See the NOTICE file
    + * distributed with this work for additional information
    + * regarding copyright ownership.  The ASF licenses this file
    + * to you under the Apache License, Version 2.0 (the
    + * "License"); you may not use this file except in compliance
    + * with the License.  You may obtain a copy of the License at
    + *
    + * http://www.apache.org/licenses/LICENSE-2.0
    + *
    + * Unless required by applicable law or agreed to in writing, software
    + * distributed under the License is distributed on an "AS IS" BASIS,
    + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
    + * See the License for the specific language governing permissions and
    + * limitations under the License.
    + */
    +
    +package org.apache.flink.streaming.api.graph;
    +
    +import com.google.common.hash.HashFunction;
    +import com.google.common.hash.Hasher;
    +import com.google.common.hash.Hashing;
    +import org.apache.flink.runtime.jobgraph.JobVertexID;
    +import org.apache.flink.streaming.api.operators.AbstractUdfStreamOperator;
    +import org.apache.flink.streaming.api.operators.ChainingStrategy;
    +import org.apache.flink.streaming.api.operators.StreamOperator;
    +import org.apache.flink.streaming.api.transformations.StreamTransformation;
    +import org.apache.flink.streaming.runtime.partitioner.ForwardPartitioner;
    +import org.slf4j.Logger;
    +import org.slf4j.LoggerFactory;
    +
    +import java.nio.charset.Charset;
    +import java.util.ArrayDeque;
    +import java.util.ArrayList;
    +import java.util.Arrays;
    +import java.util.Collections;
    +import java.util.HashMap;
    +import java.util.HashSet;
    +import java.util.List;
    +import java.util.Map;
    +import java.util.Queue;
    +import java.util.Set;
    +
    +import static org.apache.flink.util.StringUtils.byteToHexString;
    +
    +public class DefaultStreamGraphHasher implements StreamGraphHasher {
    +
    +	private static final Logger LOG = LoggerFactory.getLogger(DefaultStreamGraphHasher.class);
    +
    +	/**
    +	 * Returns a map with a hash for each {@link StreamNode} of the {@link
    +	 * StreamGraph}. The hash is used as the {@link JobVertexID} in order to
    +	 * identify nodes across job submissions if they didn't change.
    +	 * <p>
    +	 * <p>The complete {@link StreamGraph} is traversed. The hash is either
    +	 * computed from the transformation's user-specified id (see
    +	 * {@link StreamTransformation#getUid()}) or generated in a deterministic way.
    +	 * <p>
    +	 * <p>The generated hash is deterministic with respect to:
    +	 * <ul>
    +	 * <li>node-local properties (like parallelism, UDF, node ID),
    +	 * <li>chained output nodes, and
    +	 * <li>input nodes hashes
    +	 * </ul>
    +	 *
    +	 * @return A map from {@link StreamNode#id} to hash as 16-byte array.
    +	 */
    +	@Override
    +	public Map<Integer, byte[]> traverseStreamGraphAndGenerateHashes(StreamGraph streamGraph)
{
    +		// The hash function used to generate the hash
    +		final HashFunction hashFunction = Hashing.murmur3_128(0);
    +		final Map<Integer, byte[]> hashes = new HashMap<>();
    +
    +		Set<Integer> visited = new HashSet<>();
    +		Queue<StreamNode> remaining = new ArrayDeque<>();
    +
    +		// We need to make the source order deterministic. The source IDs are
    +		// not returned in the same order, which means that submitting the same
    +		// program twice might result in different traversal, which breaks the
    +		// deterministic hash assignment.
    +		List<Integer> sources = new ArrayList<>();
    +		for (Integer sourceNodeId : streamGraph.getSourceIDs()) {
    +			sources.add(sourceNodeId);
    +		}
    +		Collections.sort(sources);
    +
    +		//
    +		// Traverse the graph in a breadth-first manner. Keep in mind that
    +		// the graph is not a tree and multiple paths to nodes can exist.
    +		//
    +
    +		// Start with source nodes
    +		for (Integer sourceNodeId : sources) {
    +			remaining.add(streamGraph.getStreamNode(sourceNodeId));
    +			visited.add(sourceNodeId);
    +		}
    +
    +		StreamNode currentNode;
    +		while ((currentNode = remaining.poll()) != null) {
    +			// Generate the hash code. Because multiple path exist to each
    +			// node, we might not have all required inputs available to
    +			// generate the hash code.
    +			if (generateNodeHash(currentNode, hashFunction, hashes, streamGraph.isChainingEnabled()))
{
    +				// Add the child nodes
    +				for (StreamEdge outEdge : currentNode.getOutEdges()) {
    +					StreamNode child = outEdge.getTargetVertex();
    +
    +					if (!visited.contains(child.getId())) {
    +						remaining.add(child);
    +						visited.add(child.getId());
    +					}
    +				}
    +			} else {
    +				// We will revisit this later.
    +				visited.remove(currentNode.getId());
    +			}
    +		}
    +
    +		return hashes;
    +	}
    +
    +	/**
    +	 * Generates a hash for the node and returns whether the operation was
    +	 * successful.
    +	 *
    +	 * @param node         The node to generate the hash for
    +	 * @param hashFunction The hash function to use
    +	 * @param hashes       The current state of generated hashes
    +	 * @return <code>true</code> if the node hash has been generated.
    +	 * <code>false</code>, otherwise. If the operation is not successful, the
    +	 * hash needs be generated at a later point when all input is available.
    +	 * @throws IllegalStateException If node has user-specified hash and is
    +	 *                               intermediate node of a chain
    +	 */
    +	private boolean generateNodeHash(
    +			StreamNode node,
    +			HashFunction hashFunction,
    +			Map<Integer, byte[]> hashes,
    +			boolean isChainingEnabled) {
    +
    +		// Check for user-specified ID
    +		String userSpecifiedHash = node.getTransformationId();
    +
    +		if (userSpecifiedHash == null) {
    +			// Check that all input nodes have their hashes computed
    +			for (StreamEdge inEdge : node.getInEdges()) {
    +				// If the input node has not been visited yet, the current
    +				// node will be visited again at a later point when all input
    +				// nodes have been visited and their hashes set.
    +				if (!hashes.containsKey(inEdge.getSourceId())) {
    +					return false;
    +				}
    +			}
    +
    +			Hasher hasher = hashFunction.newHasher();
    +			byte[] hash = generateDeterministicHash(node, hasher, hashes, isChainingEnabled);
    +
    +			if (hashes.put(node.getId(), hash) != null) {
    +				// Sanity check
    +				throw new IllegalStateException("Unexpected state. Tried to add node hash " +
    +						"twice. This is probably a bug in the JobGraph generator.");
    +			}
    +
    +			return true;
    +		} else {
    +			// Check that this node is not part of a chain. This is currently
    +			// not supported, because the runtime takes the snapshots by the
    +			// operator ID of the first vertex in a chain. It's OK if the node
    +			// has chained outputs.
    +			for (StreamEdge inEdge : node.getInEdges()) {
    +				if (isChainable(inEdge, isChainingEnabled)) {
    +					throw new UnsupportedOperationException("Cannot assign user-specified hash "
    +							+ "to intermediate node in chain. This will be supported in future "
    +							+ "versions of Flink. As a work around start new chain at task "
    +							+ node.getOperatorName() + ".");
    +				}
    +			}
    +
    +			Hasher hasher = hashFunction.newHasher();
    +			byte[] hash = generateUserSpecifiedHash(node, hasher);
    +
    +			for (byte[] previousHash : hashes.values()) {
    +				if (Arrays.equals(previousHash, hash)) {
    +					throw new IllegalArgumentException("Hash collision on user-specified ID. " +
    +							"Most likely cause is a non-unique ID. Please check that all IDs " +
    +							"specified via `uid(String)` are unique.");
    +				}
    +			}
    +
    +			if (hashes.put(node.getId(), hash) != null) {
    +				// Sanity check
    +				throw new IllegalStateException("Unexpected state. Tried to add node hash " +
    +						"twice. This is probably a bug in the JobGraph generator.");
    +			}
    +
    +			return true;
    +		}
    +	}
    +
    +	/**
    +	 * Generates a hash from a user-specified ID.
    +	 */
    +	private byte[] generateUserSpecifiedHash(StreamNode node, Hasher hasher) {
    +		hasher.putString(node.getTransformationId(), Charset.forName("UTF-8"));
    +
    +		return hasher.hash().asBytes();
    +	}
    +
    +	/**
    +	 * 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;
    +	}
    +
    +	/**
    +	 * Applies the {@link Hasher} to the {@link StreamNode} (only node local
    +	 * attributes are taken into account). The hasher encapsulates the current
    +	 * state of the hash.
    +	 * <p>
    +	 * <p>The specified ID is local to this node. We cannot use the
    +	 * {@link StreamNode#id}, because it is incremented in a static counter.
    +	 * Therefore, the IDs for identical jobs will otherwise be different.
    +	 */
    +	protected void generateNodeLocalHash(StreamNode node, Hasher hasher, int id) {
    +		// This resolves conflicts for otherwise identical source nodes. BUT
    +		// the generated hash codes depend on the ordering of the nodes in the
    +		// stream graph.
    +		hasher.putInt(id);
    +
    +		if (node.getOperator() instanceof AbstractUdfStreamOperator) {
    +			String udfClassName = ((AbstractUdfStreamOperator<?, ?>) node.getOperator())
    +					.getUserFunction().getClass().getName();
    +
    +			hasher.putString(udfClassName, Charset.forName("UTF-8"));
    +		}
    +	}
    +
    +	private boolean isChainable(StreamEdge edge, boolean isChainingEnabled) {
    --- End diff --
    
    You're right. We should probably not have this for the default hasher and only have it
in the legacy hasher, then. WDYT?


> Ensure backwards compatibility of the hashes used to generate JobVertexIds
> --------------------------------------------------------------------------
>
>                 Key: FLINK-5290
>                 URL: https://issues.apache.org/jira/browse/FLINK-5290
>             Project: Flink
>          Issue Type: Sub-task
>          Components: State Backends, Checkpointing
>            Reporter: Stefan Richter
>            Assignee: Stefan Richter
>
> The way in which hashes for JobVertexIds are generated changed between Flink 1.1 and
1.2 (parallelism was considered for the hash in 1.1). We need to be backwards compatible to
old JobVertexId generation so that we can still assign state from old savepoints.



--
This message was sent by Atlassian JIRA
(v6.3.4#6332)

Mime
View raw message