-
Notifications
You must be signed in to change notification settings - Fork 707
Rosetta Code
sritchie edited this page Jan 30, 2013
·
27 revisions
A collection of MapReduce tasks translated (from Pig, Hive, MapReduce streaming, etc.) into Scalding. For fully runnable code, see the repository here.
# Emit (word, count) pairs.
def mapper
STDIN.each_line do |line|
line.split.each do |word|
puts [word, 1].join("\t")
end
end
end
# Aggregate all (word, count) pairs for a particular word.
#
# In Hadoop Streaming (unlike standard Hadoop), the reducer receives
# rows from the mapper *one at a time*, though the rows are guaranteed
# to be sorted by key (and every row associated to a particular key
# will be sent to the same reducer).
def reducer
curr_word = nil
curr_count = 0
STDIN.each_line do |line|
word, count = line.strip.split("\t")
if word != curr_word
puts [curr_word, curr_count].join("\t")
curr_word = word
curr_count = 0
end
curr_count += count.to_i
end
puts [curr_word, curr_count].join("\t") unless curr_word.nil?
end
# tokenizer.py
import sys
for line in sys.stdin:
for word in line.split():
print word
CREATE TABLE tweets (text STRING);
LOAD DATA LOCAL INPATH 'tweets.tsv' OVERWRITE INTO TABLE tweets;
SELECT word, COUNT(*) AS count
FROM (
SELECT TRANSFORM(text) USING 'python tokenizer.py' AS word
FROM tweets
) t
GROUP BY word;
tweets = LOAD 'tweets.tsv' AS (text:chararray);
words = FOREACH tweets GENERATE FLATTEN(TOKENIZE(text)) AS word;
word_groups = GROUP words BY word;
word_counts = FOREACH word_groups GENERATE group AS word, COUNT(words) AS count;
STORE word_counts INTO 'word_counts.tsv';
(cascalog.repl/bootstrap)
(defmapcatop tokenize [text] (seq (.split text "\\s+")))
(?<- (hfs-textline "output.tsv") [?word ?count]
((hfs-textline "tweets.tsv") ?textline)
(tokenize ?textline :> ?word)
(c/count ?count)))
import com.twitter.scalding._
class ScaldingTestJob(args: Args) extends Job(args) {
Tsv("tweets.tsv", 'text)
.flatMap('text -> 'word) { text : String => text.split("\\s+") }
.groupBy('word) { _.size }
.write(Tsv("word_counts.tsv"))
}
PATTERN = /.*hello.*/
# Emit words that match the pattern.
def mapper
STDIN.each_line do |line|
puts line if line =~ PATTERN
end
end
# Identity reducer.
def reducer
STDIN.each_line do |line|
puts line
end
end
%declare PATTERN '.*hello.*';
tweets = LOAD 'tweets.tsv' AS (text:chararray);
results = FILTER tweets BY (text MATCHES '$PATTERN');
(def pattern #".*hello.*")
(deffilterop matches-pattern? [text pattern]
(re-matches pattern text))
(defn distributed-grep [input pattern]
(<- [?textline]
(input ?textline)
(matches-pattern? ?textline pattern)))
(?- (stdout) (distributed-grep (hfs-textline "tweets.tsv") pattern))
val Pattern = ".*hello.*";
Tsv("tweets.tsv", 'text)
.filter('text) { text : String => text.matches(Pattern) }
# Emit (word, tweet_id) pairs.
def mapper
STDIN.each_line do |line|
tweet_id, text = line.strip.split("\t")
text.split.each do |word|
puts [word, tweet_id].join("\t")
end
end
end
# Aggregate all (word, tweet_id) pairs for a particular word.
#
# In Hadoop Streaming (unlike standard Hadoop), the reducer receives
# rows from the mapper *one at a time*, though the rows are guaranteed
# to be sorted by key (and every row associated to a particular key
# will be sent to the same reducer).
def reducer
curr_word = nil
curr_inv_index = []
STDIN.each_line do |line|
word, tweet_id = line.strip.split("\t")
if word != curr_word
# New key.
puts [curr_word, curr_inv_index.join(",")].join("\t")
curr_word = word
curr_inv_index = []
end
curr_inv_index << tweet_id
end
unless curr_word.nil?
puts [curr_word, curr_inv_index.join(", ")].join("\t")
end
end
tweets = LOAD 'tweets.tsv' AS (tweet_id:int, text:chararray);
words = FOREACH tweets GENERATE tweet_id, FLATTEN(TOKENIZE(text)) AS word;
word_groups = GROUP words BY word;
inverted_index = FOREACH word_groups GENERATE group AS word, words.tweet_id;
;; define the data
(def index [
[0 "Hello World"]
[101 "The quick brown fox jumps over the lazy dog"]
[42 "Answer to the Ultimate Question of Life, the Universe, and Everything"]
])
;; the tokenize function
(defmapcatop tokenize [text]
(seq (.split text "\\s+")))
;; ensure inverted index is distinct per word
(defbufferop distinct-vals [tuples]
(list (set (map first tuples))))
;; run the query on data
(?<- (stdout) [?word ?ids]
(index ?id ?text)
(tokenize ?text :> ?word)
(distinct-vals ?id :> ?ids))
val tweets = Tsv("tweets.tsv", ('id, 'text))
val wordToTweets =
tweets
.flatMap(('id, 'text) -> ('word, 'tweetId)) {
fields : (Long, String) =>
val (tweetId, text) = fields
text.split("\\s+").map { word => (word, tweetId) }
}
val invertedIndex =
wordToTweets.groupBy('word) { _.toList[Long]('tweetId -> 'tweetIds) }
- Scaladocs
- Getting Started
- Type-safe API Reference
- SQL to Scalding
- Building Bigger Platforms With Scalding
- Scalding Sources
- Scalding-Commons
- Rosetta Code
- Fields-based API Reference (deprecated)
- Scalding: Powerful & Concise MapReduce Programming
- Scalding lecture for UC Berkeley's Analyzing Big Data with Twitter class
- Scalding REPL with Eclipse Scala Worksheets
- Scalding with CDH3U2 in a Maven project
- Running your Scalding jobs in Eclipse
- Running your Scalding jobs in IDEA intellij
- Running Scalding jobs on EMR
- Running Scalding with HBase support: Scalding HBase wiki
- Using the distributed cache
- Unit Testing Scalding Jobs
- TDD for Scalding
- Using counters
- Scalding for the impatient
- Movie Recommendations and more in MapReduce and Scalding
- Generating Recommendations with MapReduce and Scalding
- Poker collusion detection with Mahout and Scalding
- Portfolio Management in Scalding
- Find the Fastest Growing County in US, 1969-2011, using Scalding
- Mod-4 matrix arithmetic with Scalding and Algebird
- Dean Wampler's Scalding Workshop
- Typesafe's Activator for Scalding