This tutorial assumes that you have set up Cloudera as per “cloudera quickstart vm tutorial installation” YouTube videos that you can search Google or YouTube. You can install it on VMWare (non commercial use) or on VirtualBox. I am using VMWare. Cloudera requires at least 8GB RAM and 16GB is...
Step 1: The “pom.xml” that defines the dependencies for Spark & Hadoop APIs.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 |
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>com.mytutorial</groupId> <artifactId>simple-spark</artifactId> <packaging>jar</packaging> <version>1.0-SNAPSHOT</version> <name>simple-spark</name> <url>http://maven.apache.org</url> <properties> <maven.compiler.source>1.7</maven.compiler.source> <maven.compiler.target>1.7</maven.compiler.target> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> <junit.version>4.8.1</junit.version> <hadoop.version>2.7.2</hadoop.version> <spark-version>1.3.0</spark-version> <scala.version>2.11.5</scala.version> <scala.compat.version>2.11</scala.compat.version> </properties> <dependencies> <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> <version>${junit.version}</version> <scope>test</scope> </dependency> <!-- Hadoop --> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-hdfs</artifactId> <version>${hadoop.version}</version> <exclusions> <exclusion> <groupId>javax.servlet</groupId> <artifactId>*</artifactId> </exclusion> </exclusions> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>${hadoop.version}</version> <exclusions> <exclusion> <groupId>javax.servlet</groupId> <artifactId>*</artifactId> </exclusion> </exclusions> </dependency> <!-- Apache Spark --> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-core_${scala.compat.version}</artifactId> <version>${spark-version}</version> <exclusions> <exclusion> <groupId>javax.servlet</groupId> <artifactId>*</artifactId> </exclusion> </exclusions> </dependency> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-streaming_${scala.compat.version}</artifactId> <version>${spark-version}</version> </dependency> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-streaming-flume_${scala.compat.version}</artifactId> <version>${spark-version}</version> </dependency> </dependencies> </project> |
Step 2: The Spark job that writes numbers 1 to 10 to 10 different files on HDFS.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 |
package com.mytutorial; import java.io.IOException; import java.util.Arrays; import java.util.List; import org.apache.hadoop.conf.Configuration; import org.apache.hadoop.fs.FSDataOutputStream; import org.apache.hadoop.fs.FileSystem; import org.apache.hadoop.fs.Path; import org.apache.spark.SparkConf; import org.apache.spark.api.java.JavaRDD; import org.apache.spark.api.java.JavaSparkContext; import org.apache.spark.api.java.function.VoidFunction; import com.esotericsoftware.minlog.Log; public class SimpleSparkJob { public static void main(String[] args) throws IOException { SparkConf conf = new SparkConf() .setAppName("Simple print something") .set("spark.serializer", "org.apache.spark.serializer.KryoSerializer"); JavaSparkContext context = new JavaSparkContext(conf); List<Integer> list = Arrays.asList(1, 2, 3, 4, 5, 6, 7, 8, 9, 10); JavaRDD<Integer> rdds = context.parallelize(list, 10); rdds.foreach(new VoidFunction<Integer>() { private static final long serialVersionUID = 1L; public void call(Integer arg0) throws Exception { Log.info("Value is " + arg0); Configuration conf = new Configuration(); Path filenamePath = new Path("/user/cloudera/bamboo" + System.nanoTime() + ".xml"); FileSystem fileSystem = FileSystem.get(conf); FSDataOutputStream fdos = fileSystem.create(filenamePath); fdos.writeUTF(arg0 + ""); fdos.close(); } }); context.close(); } } |
Step 3: Build the “jar” … Read more ›...
This extends Spark tutorial – writing a file from a local file system to HDFS.
This tutorial assumes that you have set up Cloudera as per “cloudera quickstart vm tutorial installation” YouTube videos that you can search Google or YouTube. You can install it on VMWare (non commercial use) or on VirtualBox.
…
This extends Spark submit – reading a file from HDFS. A SequenceFile is a flat file consisting of binary key/value pairs. It is extensively used in MapReduce as input/output formats. Like CSV, Sequence files do not store meta data, hence only schema evolution is appending new fields to the end...
Step 1: Create a simple maven Spark project using “-B” for non-interactive mode.
|
1 2 3 |
[cloudera@quickstart ~]$ mvn archetype:generate -DgroupId=com.mytutorial -DartifactId=simple-spark -B |
Step 2: Import the maven project “simple-spark” into eclipse. Step 3: The pom.xml file should have the relevant dependency jars as shown below.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 |
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>com.mytutorial</groupId> <artifactId>simple-spark</artifactId> <packaging>jar</packaging> <version>1.0-SNAPSHOT</version> <name>simple-spark</name> <url>http://maven.apache.org</url> <properties> <maven.compiler.source>1.8</maven.compiler.source> <maven.compiler.target>1.8</maven.compiler.target> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> <junit.version>4.8.1</junit.version> <hadoop.version>2.7.2</hadoop.version> <spark-version>1.3.0</spark-version> </properties> <dependencies> <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> <version>${junit.version}</version> <scope>test</scope> </dependency> <!-- Hadoop --> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-hdfs</artifactId> <version>${hadoop.version}</version> <exclusions> <exclusion> <groupId>javax.servlet</groupId> <artifactId>*</artifactId> </exclusion> </exclusions> </dependency> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-client</artifactId> <version>${hadoop.version}</version> <exclusions> <exclusion> <groupId>javax.servlet</groupId> <artifactId>*</artifactId> </exclusion> </exclusions> </dependency> <!-- Apache Spark --> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-core_2.11</artifactId> <version>${spark-version}</version> <exclusions> <exclusion> <groupId>javax.servlet</groupId> <artifactId>*</artifactId> </exclusion> </exclusions> </dependency> </dependencies> </project> |
Step 4: Write the simple Spark job “SimpleSparkJob.java” … Read more ›...
Step 1: Create a simple maven project.
|
1 2 3 |
mvn archetype:generate -B -DgroupId=com.mytutorial -DartifactId=simple-spark |
Step 2: Import the “simple-spark” maven project into eclipse or IDE of your choice. Step 3: Modify the pom.xml file include 1) relevant Spark libraries 2) The shade plugin to create a single jar (i.e. … Read more ›...
A DataFrame is an immutable distributed collection of data like an RDD, but unlike an RDD, data is organized into named columns of a table in a relational database. This makes processing easier by imposing a structure onto a distributed collection of data. From Spark 2.0 onwards, … Read more...
This extends Running a Simple Spark Job in local & cluster modes and Apache Flume with JMS source (Websphere MQ) and HDFS sink. In this tutorial a Flume sink will ingest the data from a source like JMS, HDFS, etc and pass it to an “ … Read more ›...
Processing very large XML files can be a bit tricky as they cannot be processed line by line in parallel as you would do with CSV files. The xml file has to be intact whilst matching the start and end entity tags, and if the tags are distributed in parts...
We know that the following code snippets in Spark will write each JavaRDD element to a single file
|
1 2 3 |
employeesRdd.saveAsTextFile(pathToHdfs) |
What if you want to write each employee history to a separate file? Step 1: Create a JavaPairRDD from JavaRDD
|
1 2 3 4 5 6 7 8 9 |
JavaPairRDD<String, String> empPairRdd = employeesRdd.mapToPair(new PairFunction<Employee, String, String>() { @Override public Tuple2<String, String> call(Employee t) throws Exception { return new Tuple2<String, String>(t.getEmpId, t.getEmpHistoryDump()); } }); |
Step 2: Create a MultipleOutputFormat, … Read more ›...
Step 1: The pom.xml file that has all the relevant dependencies to Spark, Avro & hadoop libraries.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 |
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>com.mytutorial</groupId> <artifactId>simple-spark</artifactId> <packaging>jar</packaging> <version>1.0-SNAPSHOT</version> <name>simple-spark</name> <url>http://maven.apache.org</url> <properties> <maven.compiler.source>1.7</maven.compiler.source> <maven.compiler.target>1.7</maven.compiler.target> <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding> <junit.version>4.8.1</junit.version> <hadoop.version>2.7.2</hadoop.version> <spark-version>1.3.0</spark-version> <avro.version>1.8.1</avro.version> </properties> <dependencies> <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> <version>${junit.version}</version> <scope>test</scope> </dependency> <!-- Hadoop API --> <dependency> <groupId>org.apache.hadoop</groupId> <artifactId>hadoop-hdfs</artifactId> <version>${hadoop.version}</version> <exclusions> <exclusion> <groupId>javax.servlet</groupId> <artifactId>*</artifactId> </exclusion> </exclusions> <scope>provided</scope> </dependency> <!-- Apache Spark --> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-core_2.11</artifactId> <version>${spark-version}</version> <exclusions> <exclusion> <groupId>javax.servlet</groupId> <artifactId>*</artifactId> </exclusion> </exclusions> <scope>provided</scope> </dependency> <!-- Apache Spark --> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-sql_2.11</artifactId> <version>${spark-version}</version> </dependency> <dependency> <groupId>com.databricks</groupId> <artifactId>spark-xml_2.11</artifactId> <version>0.4.0</version> </dependency> <!-- AVRO --> <dependency> <groupId>org.apache.avro</groupId> <artifactId>avro</artifactId> <version>${avro.version}</version> </dependency> <dependency> <groupId>org.apache.avro</groupId> <artifactId>avro-mapred</artifactId> <version>${avro.version}</version> <classifier>hadoop2</classifier> <exclusions> <exclusion> <groupId>org.apache.avro</groupId> <artifactId>*</artifactId> </exclusion> </exclusions> </dependency> </dependencies> <build> <plugins> <!-- build an uber jar with hadoop dependencies included --> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-shade-plugin</artifactId> <executions> <execution> <phase>package</phase> <goals> <goal>shade</goal> </goals> </execution> </executions> </plugin> </plugins> </build> </project> |
Step 2: Avro schema /schema/employee.avsc file under src/main/resources folder.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 |
{ "type": "record", "namespace": "com.myapp", "name": "organization", "fields": [ { "name": "name", "type": "string", "doc": "The employee name", "default": null } , { "name": "title", "type": "string", "doc": "employee title", "default": null } ] } |
Step 3: Spark job that creates random data into a RDD named “ … Read more ›...
Step 1: pom.xml with library dependencies. It is important to note that 1) “https://repository.cloudera.com/artifactory/cloudera-repos/” is added as the “Cloudera Maven Repository” and 2) hbase-spark dependency is used for writing to HBase from Spark RDDs & … Read more ›...
Using Spark streaming data can be ingested from many sources like Kafka, Flume, HDFS, Unix/Windows File system, etc. In this example, let’s run the Spark in a local mode to ingest data from a Unix file system. Step 1: The pom.xml file.
|
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 |
<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance" xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/maven-v4_0_0.xsd"> <modelVersion>4.0.0</modelVersion> <groupId>com.mytutorial</groupId> <artifactId>simple-spark</artifactId> <packaging>jar</packaging> <version>1.0-SNAPSHOT</version> <name>simple-spark</name> <url>http://maven.apache.org</url> <properties> <maven.compiler.source>1.8</maven.compiler.source> <maven.compiler.target>1.8</maven.compiler.target> <hadoop.version>2.7.2</hadoop.version> <avro.version>1.8.1</avro.version> <spark.version>1.6.0-cdh5.8.3</spark.version> <scala.version>2.10.4</scala.version> <scala.binary.version>2.10</scala.binary.version> </properties> <repositories> <repository> <id>central</id> <name>Maven Central</name> <url>http://repo1.maven.org/maven2/</url> </repository> <repository> <id>cloudera</id> <url>http://repository.cloudera.com/artifactory/cloudera-repos/</url> </repository> </repositories> <dependencies> <dependency> <groupId>junit</groupId> <artifactId>junit</artifactId> <version>4.12</version> <scope>test</scope> </dependency> <dependency> <groupId>org.scala-lang</groupId> <artifactId>scala-library</artifactId> <version>${scala.version}</version> </dependency> <!-- Spark libraries --> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-core_${scala.binary.version}</artifactId> <version>${spark.version}</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-sql_${scala.binary.version}</artifactId> <version>${spark.version}</version> <scope>provided</scope> </dependency> <dependency> <groupId>org.apache.spark</groupId> <artifactId>spark-streaming_${scala.binary.version}</artifactId> <version>${spark.version}</version> <scope>provided</scope> </dependency> </dependencies> <build> <plugins> <!-- build an uber jar with hadoop dependencies included --> <plugin> <groupId>org.apache.maven.plugins</groupId> <artifactId>maven-shade-plugin</artifactId> <executions> <execution> <phase>package</phase> <goals> <goal>shade</goal> </goals> </execution> </executions> </plugin> </plugins> </build> </project> |
Using textFileStream(..) textFileStream watches a directory for...
