Partitions the operator state of a {@link DataStream} by the given key positions.

@param fields
The position of the fields on which the {@link DataStream}
will be grouped.
@return The {@link DataStream} with partitioned state (i.e. KeyedStream)
342849
public KeyedStream<T, Tuple> keyBy(int... fields) {
		if (getType() instanceof BasicArrayTypeInfo || getType() instanceof PrimitiveArrayTypeInfo) {
			return keyBy(KeySelectorUtil.getSelectorForArray(fields, getType()));
		} else {
			return keyBy(new Keys.ExpressionKeys<>(fields, getType()));
		}
	}
918795
@TimerJ
    public TableMetadataBuilder withClusterKey(String... fields) {
        for (String field : fields) {
            clusterKey.add(new ColumnName(tableName, field));
        }
        return this;
    }
63591
public void setPartitionKeys(List<FieldSchema> partitionKeys) {
    this.setProp(HIVE_PARTITION_KEYS, GSON.toJson(partitionKeys, FIELD_SCHEMA_TYPE));
  }
629374
public SqlBuilder groupBy(String... fields) {
		if (ArrayUtil.isNotEmpty(fields)) {
			if (null != wrapper) {
				// 包装字段名
				fields = wrapper.wrap(fields);
			}

			sql.append(" GROUP BY ").append(ArrayUtil.join(fields, StrUtil.COMMA));
		}

		return this;
	}
314507
private List<Field> getList(Map<Class<?>, List<Field>> fields) {
        List<Field> ret = fields == null ? null : fields.get(getServiceClass());
        if (ret != null) {
            ret = Collections.unmodifiableList(ret);
        } else {
            ret = Collections.emptyList();
        }
        return ret;
    }
1177997
private void buildSlots(ObjectStreamField[] fields) {
        slotsToSerialize = new ObjectSlot[fields.length];
        for (int i = 0; i < fields.length; i++) {
            ObjectSlot s = new ObjectSlot();
            slotsToSerialize[i] = s;
            s.field = fields[i];
        }
        // We assume the slots are already sorted in the right shape for dumping
    }
476772
private Field getPartitionKeyField()
    {
        Field[] embeddedFields = entityMetadata.getIdAttribute().getBindableJavaType().getDeclaredFields();

        Field field = null;
        for (Field embeddedField : embeddedFields)
        {
            if (!ReflectUtils.isTransientOrStatic(embeddedField))
            {
                field = embeddedField;
                break;
            }
        }
        return field;
    }
45087
public String[] getPartitionKeys(String location, Job job) throws IOException
    {
        if (!usePartitionFilter)
            return null;
        List<ColumnDef> indexes = getIndexes();
        String[] partitionKeys = new String[indexes.size()];
        for (int i = 0; i < indexes.size(); i++)
        {
            partitionKeys[i] = new String(indexes.get(i).getName());
        }
        return partitionKeys;
    }
1360775
public byte[] getPosition(byte field) {
		Iterator it = m_fieldsPosition.keySet().iterator();
		Byte B = field;
		while (it.hasNext()) {
			Position p = (Position) it.next();
			if (((Vector) m_fieldsPosition.get(p)).contains(B)) {
				return new byte[] { p.m_vertical, p.m_horizontal };
			}
		}
		if (field != ScoreElements._DEFAULT) {
			return getPosition(ScoreElements._DEFAULT);
		}
		return null;
	}
543397
public AggregationResult aggregate(String fields) throws DDFException {
    return this.getAggregationHandler().aggregate(AggregateField.fromSqlFieldSpecs(fields));
  }
1060662
protected DataStream<T> setConnectionType(StreamPartitioner<T> partitioner) {
		return new DataStream<>(this.getExecutionEnvironment(), new PartitionTransformation<>(this.getTransformation(), partitioner));
	}
565966
private RangeAggregationBuilder rangeBuilder(MethodField field) {

        // ignore alias param
        LinkedList<KVValue> params = field.getParams().stream().filter(kv -> !"alias".equals(kv.key)).collect(Collectors.toCollection(LinkedList::new));

        String fieldName = params.poll().toString();

        double[] ds = Util.KV2DoubleArr(params);

        RangeAggregationBuilder range = AggregationBuilders.range(field.getAlias()).field(fieldName);

        for (int i = 1; i < ds.length; i++) {
            range.addRange(ds[i - 1], ds[i]);
        }

        return range;
    }
835877
public static Builder pullAll(String field, Object... values) {
        return new Builder().pullAll(field, values);
    }
1242515
def join! datastream
      left = self.data
      right = datastream.data
      columns = self.headers & datastream.headers

      raise NullJoin, 'Datastreams do not share any column' if columns.empty?

      left_groups  = Hash.new { |hash, key| hash[key] = [] }
      right_groups = Hash.new { |hash, key| hash[key] = [] }

      left.each do |row|
        values = columns.map { |column| row[column] }

        left_groups[values] << row unless values.include? nil
      end

      right.each do |row|
        values = columns.map { |column| row[column] }

        right_groups[values] << row unless values.include? nil
      end

      output = []

      left_groups.each do |key, left_group|

        if right_group = right_groups[key]

          left_group.each do |left_tuple|
            right_group.each do |right_tuple|
              output << left_tuple.merge(right_tuple)
            end
          end

        end
      end

      # TODO: I do not like this. Why a datastream's data could be
      # modified directly? Anyway, #join! is likely to be reimplemented
      # many times as more features are added. It will improve.
      self.data = output
      self
    end
1035245
public static DataPersister lookupForField(Field field) {

		// see if the any of the registered persisters are valid first
		if (registeredPersisters != null) {
			for (DataPersister persister : registeredPersisters) {
				if (persister.isValidForField(field)) {
					return persister;
				}
				// check the classes instead
				for (Class<?> clazz : persister.getAssociatedClasses()) {
					if (field.getType() == clazz) {
						return persister;
					}
				}
			}
		}

		// look it up in our built-in map by class
		DataPersister dataPersister = builtInMap.get(field.getType().getName());
		if (dataPersister != null) {
			return dataPersister;
		}

		/*
		 * Special case for enum types. We can't put this in the registered persisters because we want people to be able
		 * to override it.
		 */
		if (field.getType().isEnum()) {
			return DEFAULT_ENUM_PERSISTER;
		} else {
			/*
			 * Serializable classes return null here because we don't want them to be automatically configured for
			 * forwards compatibility with future field types that happen to be Serializable.
			 */
			return null;
		}
	}
300452
protected B fields(List<F> fields) {
        if (instance.def.fields == null) {
            instance.def.fields = new ArrayList<F>(fields.size());
        }
        instance.def.fields.addAll(fields);
        return returnThis();
    }
1114924
public Optional<List<FieldSchema>> getPartitionKeys() {
    String serialzed = this.getProp(HIVE_PARTITION_KEYS);
    if (serialzed == null) {
      return Optional.absent();
    }
    List<FieldSchema> deserialized = GSON.fromJson(serialzed, FIELD_SCHEMA_TYPE);
    return Optional.of(deserialized);
  }