def test_deterministic_key(self):
    p = df.Pipeline('DirectPipelineRunner', argv=sys.argv)
    lines = ['banana,fruit,3', 'kiwi,fruit,2', 'kiwi,fruit,2', 'zucchini,veg,3']

    # [START type_hints_deterministic_key]
    class Player(object):
      def __init__(self, team, name):
        self.team = team
        self.name = name

    class PlayerCoder(df.coders.Coder):
      def encode(self, player):
        return '%s:%s' % (player.team, player.name)

      def decode(self, s):
        return Player(*s.split(':'))

      def is_deterministic(self):
        return True

    df.coders.registry.register_coder(Player, PlayerCoder)

    def parse_player_and_score(csv):
      name, team, score = csv.split(',')
      return Player(team, name), int(score)

    totals = (
        lines
        | df.Map(parse_player_and_score)
        | df.CombinePerKey(sum).with_input_types(df.typehints.Tuple[Player, int]))
    # [END type_hints_deterministic_key]

    self.assertEquals(
        {('banana', 3), ('kiwi', 4), ('zucchini', 3)},
        set(totals | df.Map(lambda (k, v): (k.name, v))))
dataset/ETHPy150Open GoogleCloudPlatform/DataflowPythonSDK/google/cloud/dataflow/examples/snippets/snippets_test.py/TypeHintsTest.test_deterministic_key/original