Asim Jalis

Want to get new posts? Subscribe Archive Asim Resume [email protected]

[asimjalis-blog]

How to Run Scala Script on Spark

Q: How can I run a Scala script on Apache Spark without compiling it?

A: Type the script in a file. Lets say this is called myscript.scala. Make sure you put System.exit(0) at the end.

val result = sc.parallelize(Array(1,2,3,4,5)).map(x => x*x).collect() result.foreach(x => println(x)) System.exit(0)

Next, submit it using the spark-shell.

spark-shell -i myscript.scala

[asimjalis-blog]

How Spark Handles Failures

Q. In a streaming app if the receiver is getting data faster than the DStreams are able to process it what happens?

A. In this case the receiver will continue to receive data and turn them into RDDs every batch interval. In the meantime the DStream will process the RDDs at its pace. Over time the RDD queue in the DStream will build up. This is okay once in a while. But if it is happening continuously you want to fix this.

Some ways to fix this are:

Increase number of partitions.

Make the batch interval longer.

Make the window's slide duration longer.

Optimize the processing so it happens faster.

Q. How does Spark handle failure of the driver, executors, tasks in batch mode?

A. This is covered in the slides.

Failed tasks are re-attempted 4 times, like MapReduce.

Failed executors have their tasks partitions and tasks reassigned to other executors. Some partitions might need to be recomputed.

Failed drivers cause the job to fail. It needs to be restarted somewhere else. YARN and other frameworks can handle restarts.

Q. How does Spark handle failure of the driver, executors, receivers, tasks in streaming mode?

A. This is covered in a lot of depth in the Spark 1.2.0 documentation. This feature changed significantly between the different versions so you want to make sure you read 1.2.0 on this.

Here are the main sections in the docs that discuss this.

https://spark.apache.org/docs/1.2.0/streaming-programming-guide.html#fault-tolerance-semantics

https://spark.apache.org/docs/1.2.0/streaming-custom-receivers.html#receiver-reliability

Q. How do you kill a long-running Spark Streaming app?

A. Through the execution framework. For example in YARN you would say yarn application -kill APP_ID

[asimjalis-blog]

How Spark Partitions Non-HDFS Files

Q: In Spark if I create an RDD with a large local file or an S3 file (i.e. a non-HDFS file) how many partitions will the RDD have?

A: The file will be split into blocks by the Hadoop InputFormat code. By default it will assume a block size of 32M. This can be overridden using the Hadoop configuration parameter fs.local.block.size.

In org/apache/hadoop/fs/FileSystem.java the default block size is defined for all files to be 32MB.

public long getDefaultBlockSize() { // default to 32MB: large enough to minimize the impact of seeks return getConf().getLong("fs.local.block.size", 32 * 1024 * 1024); }

When you read a local file or a file from S3 using sc.textFile this is going to be its block size. Hadoop's FileInputFormat will use this to calculate how many Spark partitions (file splits) to create.

I tested this on a local file of size 1,036,923,510 bytes, and the file produced 31 partitions. This demonstrates that Spark uses a block size of 32M.

scala> val path = "file:/Users/asimjalis/d/X18-08827.dmg" path: String = file:/Users/asimjalis/d/X18-08827.dmg scala> sc.textFile(path).partitions.length res35: Int = 31 scala> 1036923510.0/(32 * 1024 * 1024) res36: Double = 30.902728736400604

[asimjalis-blog]

Apache Spark SQL

How can I use SQL to query Apache Spark RDDs?

Here are the steps. Try these out by pasting them into a Scala Spark REPL (shell).

// Create SQL context. val sqlContext = new org.apache.spark.sql.SQLContext(sc) // Create RDD containing sequence of JSON records. val jsonArray = Array( """ {"name":"Liz","address":{"city":"SF","state":"CA"}} """, """ {"name":"Bob","address":{"city":"SF","state":"CA"}} """, """ {"name":"Che","address":{"city":"LA","state":"CA"}} """) val jsonRdd = sc.parallelize(jsonArray) // Create Schema RDD and register as table. val employeesRdd = sqlContext.jsonRDD(jsonRdd) employeesRdd.registerTempTable("employees") // View the Schema RDD. employeesRdd.printSchema employeesRdd.collect // Use SQL to query the RDD. val sfEmployeesRdd = sqlContext.sql( "SELECT * FROM employees WHERE address.city = 'SF'") // This RDD will only contain the SF employees. sfEmployeesRdd.printSchema sfEmployeesRdd.collect

[asimjalis-blog]

Apache Spark and HBase

How can I use Apache Spark to query and update data stored in HBase?

This tutorial will cover how to use Apache Spark to query and update data stored HBase. Here are the steps.

Download Apache Spark from https://spark.apache.org/downloads.html. Get the latest version of Spark built for Hadoop 2.4. Then install it.

Download HBase from http://www.apache.org/dyn/closer.cgi/hbase/ and install it.

Change to the directory where you uncompressed HBase.

Create a list of all the HBase jar files.

HBASE_ROOT=$HOME/Downloads/hbase-0.98.9-hadoop2 HBASE_JARS=$(find $HBASE_ROOT -iname '*.jar' | xargs echo | sed 's/ /,/g')

Start HBase.

./bin/start-hbase.sh

Start HBase shell.

./bin/hbase shell

Change to the directory where you uncompressed Spark.

Start the Spark Scala REPL (shell).

./bin/spark-shell --master local[4] --jars $HBASE_JARS

Paste this code on the Spark Scala REPL. You can paste it incrementally, or in one shot. Make sure you modify the path to the HBase conf directory so it points to the right location on your machine.

import org.apache.spark._ import org.apache.spark.rdd.NewHadoopRDD import org.apache.hadoop.hbase.{HBaseConfiguration, HTableDescriptor} import org.apache.hadoop.hbase.client.HBaseAdmin import org.apache.hadoop.hbase.mapreduce.TableInputFormat import org.apache.hadoop.fs.Path; import org.apache.hadoop.hbase.HColumnDescriptor import org.apache.hadoop.hbase.util.Bytes import org.apache.hadoop.hbase.client.Put; import org.apache.hadoop.hbase.client.HTable; val tableName = "employees" // Add local HBase conf. val conf = HBaseConfiguration.create() val confPathString = "file:///Users/asimjalis/d/hbase-0.98.9-hadoop2/conf/hbase-site.xml" conf.addResource(new Path(confPathString)) conf.set(TableInputFormat.INPUT_TABLE, tableName) // Create table with column family. val admin = new HBaseAdmin(conf) if (!admin.isTableAvailable(tableName)) { print("Creating employees table") val tableDesc = new HTableDescriptor(tableName) tableDesc.addFamily(new HColumnDescriptor("cf1".getBytes())); admin.createTable(tableDesc) } else { throw new Exception("Table already exists. Drop it first.") } // Add data to table. val myTable = new HTable(conf, tableName); for (i <- 0 to 5) { var p = new Put(new String("row" + i).getBytes()); p.add("cf1".getBytes(), "column-1".getBytes(), new String("value " + i).getBytes()); myTable.put(p); } myTable.flushCommits(); // Create RDD. val hBaseRDD = sc.newAPIHadoopRDD(conf, classOf[TableInputFormat], classOf[org.apache.hadoop.hbase.io.ImmutableBytesWritable], classOf[org.apache.hadoop.hbase.client.Result]) // Get row count. val count = hBaseRDD.count() print("HBase RDD count:" + count)

This will create an HBase table called employees and populate it. Then it will use Spark to find out the size of the table.

To verify that the table really exists, you can view it from the HBase shell. Switch to the window running the HBase shell and type these commands.

list 'employees' scan 'employees' scan 'employees', {'LIMIT' => 2}

Exit the HBase shell. Next shut down HBase using the following command. First go to the directory where HBase is installed. Then type:

./bin/stop-hbase.sh

[asimjalis-blog]

Apache Spark and SequenceFiles

How can I read and write data in SequenceFile format from Spark?

Spark lets you both read and write SequenceFile data from local files or HDFS. In fact if you use it from Python you can use it as a convenient way to convert a SequenceFile into a non-Java format.

Here is how to write and read SequenceFiles from Spark.

Start the pyspark shell

To save a PairRDD as a SequenceFile:

rdd = sc.parallelize([('k1',1),('k2',2),('k3',3)]) rdd.saveAsSequenceFile('file:/tmp/foo')

This will create a directory at /tmp/foo containing partition files, in the SequenceFile format.

Here is how to read the SequenceFile data back into Spark.

To read the SequenceFile into Spark as an RDD:

sc.sequenceFile('file:/tmp/foo').collect()

Notes:

You can only save a PairRDD as a SequenceFile. If you try to write a non-PairRDD as a SequenceFile you will get an exception.

This feature requires Spark 1.1.1.

[asimjalis-blog]

Using Apache Spark to Find Stock High

Suppose you have stock market data from Yahoo! for AAPL from http://finance.yahoo.com/q/hp?s=AAPL+Historical+Prices. The data is in CSV format and has these values.

Date Open High Low Close Volume Adj Close ---- ---- ---- --- ----- ------ --------- 11-18-2014 113.94 115.69 113.89 115.47 44,200,300 115.47 11-17-2014 114.27 117.28 113.30 113.99 46,746,700 113.99

Here is what the CSV looks like:

csv = [ "#Date,Open,High,Low,Close,Volume,Adj Close\n", "2014-11-18,113.94,115.69,113.89,115.47,44200300,115.47\n", "2014-11-17,114.27,117.28,113.30,113.99,46746700,113.99\n", ]

Lets find the date on which the price was the highest.

Question: What two fields would you need to extract?

Date and Adj Close.

We want to use Adj Close instead of High so our calculation is not affected by stock splits.

Question: What field would need to be the key?

Adj Close.

Question: What are the sequence of operations that you would need to perform.

Use textFile to read the file.

Use filter to remove the header line.

Use map to split each row into fields.

Use map to extract Adj Close and Date.

Use sortByKey(false) to sort descending on Adj Close.

Use take(1) to get the highest value.

Here is full source.

csv = [ "#Date,Open,High,Low,Close,Volume,Adj Close\n", "2014-11-18,113.94,115.69,113.89,115.47,44200300,115.47\n", "2014-11-17,114.27,117.28,113.30,113.99,46746700,113.99\n", ] sc.parallelize(csv) \ .filter(lambda line: not line.startswith("#")) \ .map(lambda line: line.split(",")) \ .map(lambda fields: (float(fields[-1]),fields[0])) \ .sortByKey(False) \ .take(1)

Here is the program for finding the high of any stock.

import urllib2 import re def get_stock_high(symbol): url = 'http://real-chart.finance.yahoo.com' + \ '/table.csv?s='+symbol+'&g=d&ignore=.csv' csv = urllib2.urlopen(url).read() csv_lines = csv.split('\n') stock_rdd = sc.parallelize(csv_lines) \ .filter(lambda line: re.match(r'\d', line)) \ .map(lambda line: line.split(",")) \ .map(lambda fields: (float(fields[-1]),fields[0])) \ .sortByKey(False) return stock_rdd.take(1) get_stock_high('AAPL')

[asimjalis-blog]

Elephant Rider in Different Languages

In case you are looking for a project name for a Hadoop-related project and want to follow Mahout and Oozie and use a word that means elephant rider in some language, here are some ideas.

Language Elephant Rider -------- -------------- Arabic Fayal Burmese Oozie French Cornac Hindi Mahout Malayalee Paappaan Polish Kornak Portuguese Cornaca Sanskrit Karināyaka Sinhalese Kurawanayaka Tamil Pahan Thai Kwan-chang Vietnamese Quản tượng

[asimjalis-blog]

Hive and Data Conversion

How can I convert HDFS files from text to SequenceFile and other formats?

While Hive is primarily a tool for running SQL queries against data in HDFS, it can also be used as a tool to convert data from one format to another. Hive is able to do this because it has a rich collection of SerDes (serializer-deserializers) for different formats.

Here is how to convert a text file to SequenceFile format using Hive.

Create a file called sales_data with the following contents:

010,CA,Alice,100 011,NY,Bob,200 013,NY,Bob,300 013,WA,Bob,200

Start the Hive shell and type the following commands into it.

-- Create table with source format. CREATE TABLE sales_as_text ( id STRING, state STRING, name STRING, amount DOUBLE) ROW FORMAT DELIMITED FIELDS TERMINATED BY ',' LINES TERMINATED BY '\n'; -- Create table with target format. CREATE TABLE sales_as_seq ( id STRING, state STRING, name STRING, amount DOUBLE) STORED AS SEQUENCEFILE; -- Load sales data. LOAD DATA LOCAL INPATH 'sales_data' INTO TABLE sales_as_text; -- Optionally enable compression. SET hive.exec.compress.output=true; SET io.seqfile.compression.type=BLOCK; -- Use INSERT SELECT to convert formats. INSERT OVERWRITE TABLE sales_as_seq SELECT * FROM sales_as_text;

[asimjalis-blog]

Python's MrJob

How can I use Python's MrJob framework to write MapReduce jobs?

Install mrjob.

sudo easy_install pip sudo pip install mrjob

Create a test folder.

mkdir mrjob cd mrjob/

Create a simple Python word count program using vim wc.py with the following contents.

from mrjob.job import MRJob import re WORD_RE = re.compile(r"[\w']+") class MRWordFreqCount(MRJob): def mapper(self, _, line): for word in WORD_RE.findall(line): yield (word.lower(), 1) def combiner(self, word, counts): yield (word, sum(counts)) def reducer(self, word, counts): yield (word, sum(counts)) if __name__ == '__main__': MRWordFreqCount.run()

Create a simple data file called data.txt with the following contents.

hello world this is line 2 this is line 3

Here is how you can run the MapReduce program locally,

python wc.py < data.txt

To submit it into the Hadoop cluster, you have to set up HADOOP_HOME.

export HADOOP_HOME=/usr/lib/hadoop-0.20-mapreduce

Next upload the data.txt file into the cluster.

hadoop fs -put data.txt

Next submit the program with the following arguments.

python wc.py -r hadoop \ --hadoop-bin /usr/bin/hadoop \ --output-dir hdfs://localhost:8020/user/training/mrjob-output \ hdfs://localhost:8020/user/training/data.txt

This assumes that data.txt is saved at /user/training in HDFS.

For a list of all the options supported by MrJob take a look at this page https://pythonhosted.org/mrjob/guides/configs-all-runners.html.

[asimjalis-blog]

Fuse DFS

How can I mount HDFS as a local file system using Fuse DFS?

Here are the steps for doing this,

# Install hadoop-hdfs-fuse sudo yum install hadoop-hdfs-fuse # Mkdir mount point. sudo mkdir -p /mnt/hdfs # Find NameNode URI from /etc/hadoop/conf/core-site.xml # Mount HDFS using NameNode URI. sudo hadoop-fuse-dfs hdfs://0.0.0.0:8020 /mnt/hdfs # You should see HDFS under /mnt/hdfs. cd /mnt/hdfs ls -l