MaxCompute:MapReduce API

In this example, you can use the reducer as the combiner to aggregate data in the mapper to reduce shuffled data.

>>> words_df.map_reduce(mapper, reducer, combiner=reducer, group='word')

In the following example, deprecated word filtering is performed in the mapper, and the number of words in a whitelist in the reducer is plus 5.

>>> white_list_file = o.create_resource('pyodps_white_list_words', 'file', file_obj='Python\nWorld')
>>>
>>> @output(['word', 'cnt'], ['string', 'int'])
>>> def mapper(resources):
>>>     stop_words = set(r[0].strip() for r in resources[0])
>>>     def h(row):
>>>         for word in row[0].split():
>>>             if word not in stop_words:
>>>                 yield word, 1
>>>     return h
>>>
>>> @output(['word', 'cnt'], ['string', 'int'])
>>> def reducer(resources):
>>>     d = dict()
>>>     d['white_list'] = set(word.strip() for word in resources[0])
>>>     d['cnt'] = 0
>>>     def inner(keys):
>>>         d['cnt'] = 0
>>>         def h(row, done):
>>>             d['cnt'] += row.cnt
>>>             if done:
>>>                 if row.word in d['white_list']:
>>>                     d['cnt'] += 5
>>>                 yield keys.word, d['cnt']
>>>         return h
>>>     return inner
>>>
>>> words_df.map_reduce(mapper, reducer, group='word',
>>>                     mapper_resources=[stop_words], reducer_resources=[white_list_file])
     word  cnt
0   hello    2
1    life    1
2  python    7
3   world    6
4   short    1
5     use    1

The usage is similar to that of a third-party Python library in map stages.

• Specify the library globally:

  >>> from odps import options
  >>> options.df.libraries = ['six.whl', 'python_dateutil.whl']

• If you use an immediately-invoked method, specify the library locally:

  >>> df.map_reduce(mapper=my_mapper, reducer=my_reducer, group='key').execute(libraries=['six.whl', 'python_dateutil.whl'])

  Note Due to the difference in bytecode definitions, if you write code based on new features supported by Python 3, such as yield from, an error is reported when the code is executed on a MaxCompute worker of Python 2.7. Before you implement production operations by using the MapReduce API of Python 3, we recommend that you make sure that the code runs normally.

If data is unevenly distributed in a cluster, you can call the reshuffle method to reshuffle data.

>>> df1 = df.reshuffle()

By default, data is hashed as random numbers. You can also distribute data by column and sort the reshuffled data in a specific order.

>>> df1.reshuffle('name', sort='id', ascending=False)

If you specify a collection object and its sequence1 for column calculation, you can apply the Bloom filter to sequence2. This way, the data that exists in sequence1 rather than sequence2 is filtered out, but the data that is not in sequence2 may not be completely filtered out. This is an approximate method. This method quickly filters out useless data from the collection object. The Bloom filter is suitable for JOIN operations in the scenario where data volumes are significantly different. Most data is not joined in this scenario. For example, most user visits do not create transactions. If you want to join visit data with transaction data, you can apply the Bloom filter to visit data before you join visit data with transaction data. This enhances operation performance.

>>> df1 = DataFrame(pd.DataFrame({'a': ['name1', 'name2', 'name3', 'name1'], 'b': [1, 2, 3, 4]}))
>>> df1
       a  b
0  name1  1
1  name2  2
2  name3  3
3  name1  4
>>> df2 = DataFrame(pd.DataFrame({'a': ['name1']}))
>>> df2
       a
0  name1
>>> df1.bloom_filter('a', df2.a) # The row 0 can be a computation expression, for example, df1.a + '1'.
       a  b
0  name1  1
1  name1  4

PyODPS DataFrame provides the pivot table feature. The following code shows the sample table data:

>>> df
     A    B      C  D  E
0  foo  one  small  1  3
1  foo  one  large  2  4
2  foo  one  large  2  5
3  foo  two  small  3  6
4  foo  two  small  3  4
5  bar  one  large  4  5
6  bar  one  small  5  3
7  bar  two  small  6  2
8  bar  two  large  7  1

DataFrame can extract key-value pairs into columns, and convert standard columns to key-value pairs. For more information about how to create and manage a DataFrame object, see Create a DataFrame object.

• Extract key-value pairs into columns. Example:

  >>> df
      name               kv
  0  name1  k1=1,k2=3,k5=10
  1  name1    k1=7.1,k7=8.2
  2  name2    k2=1.2,k3=1.5
  3  name2      k9=1.1,k2=1

  Use the extract_kv method to extract key-value fields:

  >>> df.extract_kv(columns=['kv'], kv_delim='=', item_delim=',')
     name   kv_k1  kv_k2  kv_k3  kv_k5  kv_k7  kv_k9
  0  name1    1.0    3.0    NaN   10.0    NaN    NaN
  1  name1    7.0    NaN    NaN    NaN    8.2    NaN
  2  name2    NaN    1.2    1.5    NaN    NaN    NaN
  3  name2    NaN    1.0    NaN    NaN    NaN    1.1

  The columns parameter specifies the fields that you want to extract. The kv_delim parameter specifies the delimiter between the keys and values. The item_delim parameter specifies the delimiter between key-value pairs. If you do not specify the parameters, keys and values are separated with colons (:), and key-value pairs are separated with commas (,). The name of the output field is the combination of the original field name and key value. The name and key value are connected by using an underscore (_). The default value for a missing column is NONE. You can use fill_value to fill the values for missing columns.

  >>> df.extract_kv(columns=['kv'], kv_delim='=', fill_value=0)
     name   kv_k1  kv_k2  kv_k3  kv_k5  kv_k7  kv_k9
  0  name1    1.0    3.0    0.0   10.0    0.0    0.0
  1  name1    7.0    0.0    0.0    0.0    8.2    0.0
  2  name2    0.0    1.2    1.5    0.0    0.0    0.0
  3  name2    0.0    1.0    0.0    0.0    0.0    1.1

• Convert multiple columns to key-value pairs. Example:

  >>> df
     name    k1   k2   k3    k5   k7   k9
  0  name1  1.0  3.0  NaN  10.0  NaN  NaN
  1  name1  7.0  NaN  NaN   NaN  8.2  NaN
  2  name2  NaN  1.2  1.5   NaN  NaN  NaN
  3  name2  NaN  1.0  NaN   NaN  NaN  1.1

  Use the to_kv method to convert data to the key-value format:

  >>> df.to_kv(columns=['k1', 'k2', 'k3', 'k5', 'k7', 'k9'], kv_delim='=')
      name               kv
  0  name1  k1=1,k2=3,k5=10
  1  name1    k1=7.1,k7=8.2
  2  name2    k2=1.2,k3=1.5
  3  name2      k9=1.1,k2=1