It creates a new {@link KeyedStream} that uses the provided key for partitioning
its operator states.

@param key
The KeySelector to be used for extracting the key for partitioning
@return The {@link DataStream} with partitioned state (i.e. KeyedStream)
760530
public <K> KeyedStream<T, K> keyBy(KeySelector<T, K> key) {
		Preconditions.checkNotNull(key);
		return new KeyedStream<>(this, clean(key));
	}
1248951
public static <K, V> KeyValue<K, V> empty(K key) {
        return new KeyValue<K, V>(key, null);
    }
205238
private static byte[] convertKeyToByteArray(String key) {
        byte[] result = new byte[key.length()];
        for (int i = 0; i < result.length; i++) {
            char thisChar = key.charAt(i);
            result[i] = (byte) (thisChar >>> 8 & 0xFF ^ thisChar & 0xFF);
        }
        return result;
    }
1285827
public int addKey(String key) {
        JdkUtils.requireNonNull(key);
        int nextIndex = keys.size();
        final Integer mapIndex = JdkUtils.putIfAbsent(keys, key, nextIndex);
        return mapIndex == null ? nextIndex : mapIndex;
    }
174452
public PythonKeyedStream key_by(KeySelector<PyObject, PyKey> selector) throws IOException {
		return new PythonKeyedStream(stream.keyBy(new PythonKeySelector(selector)));
	}
1227475
private Integer getKeyPartitionId(byte[] key) {
        Integer keyPartitionId = storeInstance.getNodesPartitionIdForKey(nodeId, key);

        Utils.notNull(keyPartitionId);
        return keyPartitionId;
    }
420654
private ByteBuffer encodeAndAllocateBuffer(String key) {
        if (key == null) {
            return ByteBuffer.wrap(EMPTY);
        }

        CharsetEncoder encoder = CharsetUtil.encoder(charset);
        ByteBuffer buffer = ByteBuffer.allocate((int) (encoder.maxBytesPerChar() * key.length()));

        ByteBuf byteBuf = Unpooled.wrappedBuffer(buffer);
        byteBuf.clear();
        encode(key, byteBuf);
        buffer.limit(byteBuf.writerIndex());

        return buffer;
    }
500912
public static byte[] xorBytes (byte[] data, byte[] key)
    {
        byte[] xored = new byte[data.length];
        for (int ii = 0; ii < data.length; ii++) {
            xored[ii] = (byte)(data[ii] ^ key[ii % key.length]);
        }
        return xored;
    }
959331
public Stream<T> headStream(T key, boolean inclusive, boolean parallel)
    {
        return StreamSupport.stream(headSpliterator(key, inclusive), parallel);
    }
1340819
private boolean _createDeltas(Serializable key)
    {
        if (isInitialStateMarked())
        {
            if (_copyFullInitialState && _initialFullState == null)
            {
                if (_initialState == null)
                {
                    // Copy it directly
                    _initialFullState = new HashMap<Serializable, Object>();
                    copyMap(_component.getFacesContext(), _fullState, _initialFullState);
                }
                else
                {
                    // Create only if the passed key is not part of the defined initial state
                    boolean keyInInitialState = false;
                    for (int i = 0; i < _initialState.length; i+=2)
                    {
                        Serializable key2 = (Serializable) _initialState[i];
                        if (key.equals(key2))
                        {
                            keyInInitialState = true;
                            break;
                        }
                    }
                    if (!keyInInitialState)
                    {
                        // Copy it directly, but note in this case if the initialFullState map
                        // contains some key already defined in initialState, this key must be
                        // overriden. It is better to do in that way, because it is possible
                        // to skip resetState() if the view cannot be recycled.
                        _initialFullState = new HashMap<Serializable, Object>();
                        copyMap(_component.getFacesContext(), _fullState, _initialFullState);
                        /*
                        for (int i = 0; i < _initialState.length; i+=2)
                        {
                            Serializable key2 = (Serializable) _initialState[i];
                            Object defaultValue = _initialState[i+1];
                            _initialFullState.put(key2, defaultValue);
                        }*/
                    }
                }
            }
            if (_deltas == null)
            {
                _deltas = new HashMap<Serializable, Object>(2);
            }
            return true;
        }

        return false;
    }
326647
protected void accountForFetchedKey(byte[] key) {
        fetched++;
        if (streamStats != null) {
            streamStats.reportStreamingFetch(operation);
        }

        if (recordsPerPartition <= 0) {
            return;
        }

        Integer keyPartitionId = getKeyPartitionId(key);
        Long partitionFetch = partitionFetches.get(keyPartitionId);
        Utils.notNull(partitionFetch);
        partitionFetch++;

        partitionFetches.put(keyPartitionId, partitionFetch);
        if (partitionFetch == recordsPerPartition) {
            if (partitionsToFetch.contains(keyPartitionId)) {
                partitionsToFetch.remove(keyPartitionId);
            } else {
                logger.warn("Partitions to fetch did not contain expected partition ID: "
                        + keyPartitionId);
            }
        } else if (partitionFetch > recordsPerPartition) {
            logger.warn("Partition fetch count larger than expected for partition ID "
                    + keyPartitionId + " : " + partitionFetch);
        }
    }
1161704
@Override
    public V get(K key) {
        if(key == null) {
            return null;
        }
        
        byte[] bytes = _store.get(_keySerializer.serialize(key));
        return bytes == null ? null : _valSerializer.deserialize(bytes);
    }
247048
@Override
  public Deque<MutablePair<DeltaType, Object>> getByKey(String key) {
    lock.readLock().lock();
    try {
      Deque<MutablePair<DeltaType, Object>> deltas = this.items.get(key);
      if (deltas != null) {
        // returning a shallow copy
        return new LinkedList<>(deltas);
      }
    } finally {
      lock.readLock().unlock();
    }
    return null;
  }
546528
final public Vector vector(K key)
    {
        Vector vector = storage.get(key);
        if (vector == null) return null;
        return vector;
    }
192422
public byte[] get(byte[] key){
		byte[] value = null;
		Jedis jedis = jedisPool.getResource();
		try{
			value = jedis.get(key);
		}finally{
			jedisPool.returnResource(jedis);
		}
		return value;
	}
445453
public Set<Object> hkeys(Object key) {
		Jedis jedis = getJedis();
		try {
			Set<byte[]> fieldSet = jedis.hkeys(keyToBytes(key));
			Set<Object> result = new HashSet<Object>();
			fieldSetFromBytesSet(fieldSet, result);
			return result;
		}
		finally {close(jedis);}
	}
737746
public Data get(Key key) {
		synchronized (cache) {
			Pair<Data,Long> p = cache.get(key);
			if (p != null) {
				p.setValue2(Long.valueOf(System.currentTimeMillis()));
				return p.getValue1();
			}
			p = new Pair<>(provider.provide(key), Long.valueOf(System.currentTimeMillis()));
			cache.put(key, p);
			return p.getValue1();
		}
	}