View source for Running Spark

==Loading a text file==
Save these lines into a file called ''triples.txt'' in some folder, for example your home folder ''~'':
 <nowiki>
AbrahamAbel born 1992
AbrahamAbel livesIn duluth
AbrahamAbel phone 789456123
BetsyBertram born 1987
BetsyBertram livesIn berlin
BetsyBertram phone _anon001
BetsyBertram phone _anon002
_anon001 area 56
_anon001 local 9874321
_anon001 ext 123
_anon002 area 56
_anon002 local 1234789
CharleneCharade born 1963
CharleneCharade bornIn bristol
CharleneCharade address _anon003
CharleneCharade knows BetsyBertram
_anon003 street brislingtonGrove
_anon003 number 13
_anon003 postCode bs4
_anon003 postName bristol
DirkDental born 1996
DirkDental bornIn bergen
DirkDental knows BetsyBertram
DirkDental knows CharleneCharade</nowiki>

In a console (or command prompt, or terminal) window, start the Spark shell:
    spark-shell

From now on, we will run our commands inside the Spark shell, after the ''scala> '' prompt. Load the ''triples.txt'' file into Spark:
    val triples_str = sc.textFile("/home/sinoa/triples.txt")
(You must use a forwards-slash: / even on Windows.)

''triples_str'' is now the name of a Resilient Distributed Dataset inside your spark-shell. You can enforce file loading and look at the resulting contents of the ''triples_str'' RDD with:
    triples_str.collect()
(In Scala, you can always drop empty parentheses: (), which we will do from now - so ''triples.collect'' also works.)

===Spark transformations and actions===
We are now ready to try out simple Spark transformations and actions: transformations create new RDDs when they are run, whereas actions produce side-effects or simpler variables. 

This action counts the number of lines in ''triples.txt'' (or strings in ''triples_str''):
    triples_str.count
These actions get the first and 5 first lines in ''triples_str'':
    triples_str.first
    triples_str.take(5)
This action saves the triples into a subfolder of ''/home/sinoa/triples_copy'':
    triples_str.saveAsTextFile("/home/sinoa/triples_copy")

This transformation creates a new RDD with a sample of non-duplicate lines from ''triples.txt''.
    triples_str.sample(false, 0.5, scala.util.Random.nextInt).collect
This transformation is likely to introduce duplicate lines:
    triples_str.sample(true, 0.9, scala.util.Random.nextInt).collect
Save the result in a new RDD and rerun until ''truiples_dup'' contains at least one duplicate line:
    val triples_dup = triples_str.sample(true, 0.9, scala.util.Random.nextInt)
    triples_dup.collect
This transformation removes duplicates:
    triples_dup.distinct.collect

These are only a few of the simplest Spark transformations and actions. For a full list, see this tutorial page: https://www.tutorialspoint.com/apache_spark/apache_spark_core_programming.htm .

===Unions and intersections===
Save these lines into a file called ''more_triples.txt'':
 <nowiki>
DirkDental born 1996
DirkDental bornIn bergen
DirkDental knows CharleneCharade
EnyaEntity born 2002
EnyaEntity address _anon001
EnyaEntity knows CharleneCharade
EnyaEntity knows DirkDental
_anon001 street emmastrasse
_anon001 number 7
_anon001 postArea _anon002
_anon002 postCode 45130
_anon002 postName Essen</nowiki>

These transformation produces all the lines that are in both files and all the lines that are in either file:
    val more_triples = sc.textFile("/home/sinoa/more_triples.txt")
    triples_str.union(more_triples).collect
    triples_str.intersection(more_triples).collect

===Functions===
One of the most important ideas in Spark, is passing anonymous functions as parameters to Spark transformations and actions. The functions have to be written using Scala syntax. We cannot go into detail about Scala's anonymous function syntax, but we can go some way using very simple functions (although eventually you will need a deeper understanding of Java's and Scala's overlapping type systems).

This action concatenates all the lines in "triples_str" into a single string:
    triples_str.reduce(_ + _)
Here, we passed the built-in function + (using the underscores _ to indicate the position of the two parameters to +).

If we want more control, we can write the action like this:
    triples_str.reduce((str1, str2) => str1 + " // " + str2)
Here, we defined our own anonymous Scala function ''(str1, str2) => str1 + " // " + str2'', with two input parameters ''str1'' and ''str2'' and the concatenated string ''str1 + " // " + str2'' as the result (Spark and Scala work together to keep track of the parameter and result types for us).

We can also use boolean anonymous functions to filter (a transformation) lines in our RDD:
    triples_str.filter(line => line.matches("AbrahamAbel.*")).collect
Here, the anonymous Scala function is ''line => line.matches("AbrahamAbel.*")'' with ''line'' as parameter and ''line.matches("AbrahamAbel.*")'' as the result. The latter expression is very similar to the one we wrote in Java for Hadoop: Scala builds on Java and on Java's String API.

This transformation splits each line (a string) in ''triples_str'' into an array of three strings:
    val triples = triples_str.map(line => line.split(" "))
This transformation creates a list of all the subjects, predicates, and objects in the triples:
    triples.flatMap(arr => Array(arr(0), arr(1), arr(2))).collect
We can now map the triples back to nicer-looking strings:
    triples.map(arr => "<" + arr(0) + " " + arr(1) + " " + arr(2) + "> ").collect
    triples.map(arr => "<" + arr(0) + " " + arr(1) + " " + arr(2) + "> ").reduce(_+_)
(In Scala, ''arr(0)'' is the first element in an array ''arr'', ''arr(1)'' is the second, as so on.)

You can map the set of triples by their subject (setting ''arr(0)'' to be the ''key'') and then reduce them by subject as follows:
    val triples_map = triples.map(arr => (arr(0), "<" + arr(0) + " " + arr(1) + " " + arr(2) + "> "))
    triples_map.reduceByKey(_+_)

==Tasks==
# Use Spark's ''flatMap'' transformation to collect an array of distinct resources from the triples (i.e., those strings starting with a capital letter).
# After reduction, the triple ''<CharleneCharade knows BetsyBertram>'' only appears for ''CharleneCharade''. We want it to appear for ''BetsyBertram'' too.
# After reduction, the triple ''<_anon001 area 56>'' appears for ''_anon001''. We want to eliminate ''_anon001'' so that it appears for ''BetsyBertram'' instead. The trick is to use Spark's ''join'' transformation in the right way.
# Include the triples from ''more_triples.txt'' in the map-reduce too. Note that ''_anon001'' occurs in both files, but represents a ''different'' anonymous node.
# Make sure that your map-reduce job also eliminates ''nested'' anonymouse nodes: ''more_triples.txt'' has two levels of anonymouse nodes, so that the triple ''<_anon002 postCode 555>'' appears for ''EnyaEntity''.

==Solutions to the Tasks==