How to process Instagram images with Spark 1.1.0 and OpenCV (with Java)

by Ruben Tous

Installation: You will need

Apache Spark 1.1.0
OpenCV 2.4.9 or 3.0.0 (you can follow this guide)
Apache Maven

Preparing the data

We will have thousands of .jpg files, along with their corresponding .txt files (with json metadata). On the one hand, it is not convenient to store all the files within the same directory. On the other hand, it is not convenient to access the files one by one from Spark. So, we will need to perform some previous steps:

When capturing the images it is better to save them using a hierarchy of directories (e.g. with max 1000 files for directory). If you already have the images within a big directory you can try directly performing the next steps or you can split the directory with this script
Pack all the .jpg files from a directory (500 files) into a tarball. Pack all the .txt files (500 files) into a different tarball

Transform each tarball in a Hadoop SequenceFile with this tool by Stuart Sierra

Building the code

Create the following directory structure (let's call the root YOUR_APP_PATH):

Edit the pom.xml file:

<![CDATA[<project>
  <groupId>edu.berkeley</groupId>
  <artifactId>simple-project</artifactId>
  <modelVersion>4.0.0</modelVersion>
  <name>Simple Project</name>
  <packaging>jar</packaging>
  <version>1.0</version>

<dependencies>
    <dependency> 
      <groupId>org.apache.spark</groupId>
      <artifactId>spark-core_2.10</artifactId>
      <version>1.1.0</version>
      <scope>provided</scope>
    </dependency>

<dependency>
        <groupId>opencv</groupId>
        <artifactId>opencv</artifactId>
        <version>2.4.9</version>
    </dependency>

<dependency>
        <groupId>com.googlecode.json-simple</groupId>
        <artifactId>json-simple</artifactId>
        <version>1.1</version>
    </dependency>
                 
  </dependencies>

<build>    
    <plugins>
      <plugin>       
        <artifactId>maven-assembly-plugin</artifactId>
        <version>2.4.1</version>
        <configuration>
          <descriptorRefs>
            <descriptorRef>jar-with-dependencies</descriptorRef>
          </descriptorRefs>
      <archive>
            <manifest>
              <mainClass>bigpicture.Main</mainClass>
            </manifest>
          </archive>
        </configuration>
     </plugin>
    </plugins>
   </build>
</project>]]>

This pom.xml uses the maven-assembly-plugin to pack all the dependent jars (except for the Spark-related jars) within a single .jar. This will make easier the submission of the application to Spark.

Now edit Main.java. This class 1) reads the images from the SequenceFile into a key-value RDD (key will be the filename) and uses OpenCV to determine the width of the images, 2) reads the metadata from the SequenceFile into another key-value RDD, 3) joins both RDD into a single RDD where the value is a tuple (image width, username)

<![CDATA[package bigpicture;

import org.apache.spark.api.java.*;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.function.Function;
import org.apache.hadoop.io.*;
import scala.Tuple2;
import java.util.List;
import org.apache.spark.api.java.function.PairFunction;
import org.opencv.core.Core;
import org.opencv.core.Mat;
//import org.opencv.imgcodecs.Imgcodecs; (opencv 3.0) 
import org.opencv.highgui.*; //opencv 2.4.9    
import org.opencv.utils.Converters;
import java.util.Arrays;
import org.apache.commons.io.FilenameUtils;
import org.json.simple.*;
import org.json.simple.parser.JSONParser;

public class Main {

static{ System.loadLibrary(Core.NATIVE_LIBRARY_NAME); }

public static void main(String[] args) {       
    String imagesSequenceFile = "/media/disk2/dir_1.hsf"; //change this
    String metadataSequenceFile = "/media/disk2/dir_1_meta.hsf";  //change this
    SparkConf conf = new SparkConf().setAppName("SequenceFile Test");        
    JavaSparkContext sc = new JavaSparkContext(conf);       
        
    //1) get a non serializable RDD with the images
    JavaPairRDD<Text,BytesWritable> imagesRDD_ = sc.sequenceFile(imagesSequenceFile, Text.class, BytesWritable.class);         
        
    //2) get a serializable RDD with the information extracted from the images
    JavaPairRDD<String, String> imagesRDD = imagesRDD_.mapToPair(
        new PairFunction<Tuple2<Text,BytesWritable>,String,String>() {
            @Override
            public Tuple2<String, String> call(Tuple2<Text, BytesWritable> kv) throws Exception {   
                String filename = kv._1.toString();
                String filenameWithoutExtension = FilenameUtils.getName(filename);      
                Mat image = byteswritableToOpenCVMat(kv._2);        
                return new Tuple2(filenameWithoutExtension, image.cols()+"");
            }
        });

//3) get a non serializable RDD with the metadata
    JavaPairRDD<Text,BytesWritable> metadataRDD_ = sc.sequenceFile(metadataSequenceFile, Text.class, BytesWritable.class);
    JavaPairRDD<String, String> metadataRDD = metadataRDD_.mapToPair(new PairFunction<Tuple2<Text,BytesWritable>,String,String>() {
        @Override
        public Tuple2<String, String> call(Tuple2<Text, BytesWritable> kv) throws Exception {        
            String metadataString = byteswritableToString(kv._2);

String userName = "";
            try {
                System.out.println("parsing:"+metadataString);
                JSONParser parser = new JSONParser();
                JSONObject jsonObject  = (JSONObject) parser.parse(metadataString.trim());
                JSONObject userObj = (JSONObject) jsonObject.get("user");                    
                userName = (String) userObj.get("username");                                    
            } catch (Exception e) {
                System.out.println("error parsing file "+kv._1.toString()+": "+e.getMessage());
                e.printStackTrace();
            }
            String filename = kv._1.toString();
            String filenameWithoutExtension = FilenameUtils.getBaseName(filename);                         
            return new Tuple2(filenameWithoutExtension, userName);
        }
    });               
    //4) join both RDDs
     JavaPairRDD<String, Tuple2<String, String>> joinRDD = imagesRDD.join(metadataRDD);                 
     List<Tuple2<String, Tuple2<String, String>>> output3 = joinRDD.collect();
     for (Tuple2<String,Tuple2<String, String>> tuple : output3) {
         System.out.println(tuple._1() + ": "+tuple._2()._1() + ": "+tuple._2()._2());          
     }         
  }

/* HELPER FUNCTIONS */
    public static Mat byteswritableToOpenCVMat(BytesWritable inputBW) {
        byte[] imageFileBytes = inputBW.getBytes();
        Mat img = new Mat();
        Byte[] bigByteArray = new Byte[imageFileBytes.length];
        for (int i=0; i<imageFileBytes.length; i++)                        
            bigByteArray[i] = new Byte(imageFileBytes[i]);            
        List<Byte> matlist = Arrays.asList(bigByteArray);        
        img = Converters.vector_char_to_Mat(matlist);
        //img = Imgcodecs.imdecode(img, Imgcodecs.CV_LOAD_IMAGE_COLOR); OPENCV 3.0
        img = Highgui.imdecode(img, Highgui.CV_LOAD_IMAGE_COLOR);         
        return img;
    }

public static String byteswritableToString(BytesWritable inputBW) {
        byte[] imageFileBytes = inputBW.getBytes();
        String metadataString = new String(imageFileBytes, 0, imageFileBytes.length, java.nio.charset.Charset.forName("ISO-8859-1"));        
        return metadataString;
    }
}
]]>

Before building it is necessary to import the OpenCV .jar (opencv-249.jar or opencv-300.jar) into the local Maven respository. Type the following:

In order to build, simply type:

Execute on Spark 1.1.0

Type the following from

The previous command tells Spark where to find the native OpenCV library. However, if you copy the library in /usr/lib/ it would work without the --driver-library-path directive.