apache/beam · error · ValueError

Invalid value " " for "equalities". It should be a str or a…

Error message

Invalid value "{equalities}" for "equalities". It should be a str or a list.

What it means

Thrown by _validate_equalities in apache_beam/yaml/yaml_join.py when the 'equalities' parameter is neither a str (a shared field name) nor a list (list of per-equality dicts). Any other type is rejected.

Solutions

Wrap the mapping in a list: equalities: [{input1: id, input2: id}].
Or use the plain string shorthand when the field name is identical in all inputs.
Validate the value is str or list of dicts before invoking the transform.

Example fix

// before
equalities: {input1: id, input2: id}
// after
equalities:
  - input1: id
    input2: id

Defensive patterns

Strategy: validation

Validate before calling

if not (isinstance(eq, str) or isinstance(eq, list)):
    raise TypeError("'equalities' must be a str or a list of dicts")

Type guard

def is_valid_equalities(eq):
    return isinstance(eq, str) or (isinstance(eq, list) and all(isinstance(e, dict) for e in eq))

Prevention

Remember the dict form must be wrapped in a list
Use the plain string shorthand when all inputs share the key
Validate the transform config against the Join schema before running

When it happens

Trigger: equalities: given as a dict directly ({input1: id, input2: id}) instead of wrapped in a list; equalities given as a number or null.

Common situations: YAML authors writing the mapping form without the outer list; programmatic callers passing a single dict.

Understand the failure class

Background: "Invalid ... format", "must be in format X", "does not look like a ..." — invalid argument format errors across CLI tools and libraries — this error's family across 17 libraries.

Related errors

AI-assisted analysis of apache/beam@12126d8942 (2026-09-13). Data as JSON: /api/errors/7cb56df7a5e2fd43. Report an issue: GitHub.

Appendix: source

Thrown at sdks/python/apache_beam/yaml/yaml_join.py:80

  valid_cols = {
      name: set(
          dict(fields).keys() if fields and all(
              isinstance(field, tuple) for field in fields) else fields)
      for (name, pcoll) in pcolls.items()
      for fields in [getattr(pcoll.element_type, '_fields', [])]
  }

  if isinstance(equalities, str):
    for cols in valid_cols.values():
      if equalities not in cols:
        raise ValueError(
            f'{error_prefix} When "equalities" is a str, '
            f'it must be a field name that exists in all the specified inputs.')
    equality = {pcoll_tag: equalities for pcoll_tag in pcolls}
    return [equality]

  if not isinstance(equalities, list):
    raise ValueError(f'{error_prefix} It should be a str or a list.')

  input_edge_list = []
  for equality in equalities:
    invalid_dict_error = ValueError(
        f'{error_prefix} {equality} '
        f'should be a dict[str, str] containing at least 2 items.')
    if not isinstance(equality, dict):
      raise invalid_dict_error
    if len(equality) < 2:
      raise invalid_dict_error

    for pcoll_tag, col in equality.items():
      if pcoll_tag not in pcolls:
        raise ValueError(
            f'{error_prefix} "{pcoll_tag}" is not a specified alias in "input"')
      if col not in valid_cols[pcoll_tag]:
        raise ValueError(
            f'{error_prefix} "{col}" is not a valid field in "{pcoll_tag}".')

View on GitHub (pinned to 12126d8942)