Mapreduce katsetused Azure platvormil

MapReduce Azure platvormil

Microsofti "Getting started" artikkel

Katsetustes kasutame shelli, Pythonit ja stream-I/O-d, mis ei eelda mingite pakettide ja arenduskeskkondade installimist.

Microsofti Pythoni tutorial

hdfs käsud

Katsetus: word count

mapper.py

#!/usr/bin/env python

import sys

# 'file' in this case is STDIN
def read_input(file):
    # Split each line into words
    for line in file:
        yield line.split()

def main(separator='\t'):
    data = read_input(sys.stdin)
    # Process each words returned from read_input
    for words in data:
        # Process each word
        for word in words:
            # Write to STDOUT
            print('%s%s%d' % (word, separator, 1))

if __name__ == "__main__":
    main()

reducer.py

#!/usr/bin/env python

from itertools import groupby
from operator import itemgetter
import sys

# 'file' in this case is STDIN
def read_mapper_output(file, separator='\t'):
    # Go through each line
    for line in file:
        # Strip out the separator character
        yield line.rstrip().split(separator, 1)

def main(separator='\t'):
    data = read_mapper_output(sys.stdin, separator=separator)
    # Group words and counts into 'group'
    #   Since MapReduce is a distributed process, each word
    #   may have multiple counts. 'group' will have all counts
    #   which can be retrieved using the word as the key.
    for current_word, group in groupby(data, itemgetter(0)):
        try:
            # For each word, pull the count(s) for the word
            #   from 'group' and create a total count
            total_count = sum(int(count) for current_word, count in group)
            # Write to stdout
            print("%s%s%d" % (current_word, separator, total_count))
        except ValueError:
            # Count was not a number, so do nothing
            pass

if __name__ == "__main__":
    main()

Mingid sisendfailid on juba olemas:

 hdfs dfs -ls /example/data/

Tõmbame nende pealt käima, /example on kirjutatav storage

 yarn jar /usr/hdp/current/hadoop-mapreduce-client/hadoop-streaming.jar -files mapper.py,reducer.py -mapper mapper.py -reducer reducer.py -input wasbs:///example/data/gutenberg/davinci.txt -output wasbs:///example/wordcountout

Väljund

 hdfs dfs -text /example/wordcountout/part-00000 | head -50

Katsetus: sort

mapper.py

#!/usr/bin/env python

import sys

# 'file' in this case is STDIN
def read_input(file):
    for line in file:
        yield (x.strip() for x in line.split(","))

def main(separator='\t'):
    data = read_input(sys.stdin)
    # For movielens ratings.csv
    for userid, movieid, rating, ts in data:
        hhash = "%s|%s"%(rating, movieid)
        rest = "%s|%s"%(userid, ts)
        print('%s%s%s' % (hhash, separator, rest))

if __name__ == "__main__":
    main()

reducer.py

#!/usr/bin/env python

#from itertools import groupby
#from operator import itemgetter
import sys

# 'file' in this case is STDIN
def read_mapper_output(file, separator='\t'):
    for line in file:
        yield line.rstrip().split(separator, 1)

def main(separator='\t'):
    data = read_mapper_output(sys.stdin, separator=separator)
    # just copy output (assuming it might be fed to other
    # reducers)
    for hhash, rest in data:
        print("%s%s%s" % (hhash, separator, rest))

if __name__ == "__main__":
    main()

Testandmed: http://files.grouplens.org/datasets/movielens/ml-20m.zip

Kopeeri hdfs-i:

 hdfs dfs -put ml-20m/ratings.csv /example/data/
 

Katsetus: otsing

mapper.py

reducer.py