Apache spark テキスト読み込みテスト (JUnit)

#### 1. はじめに
Apache Sparkは巨大なデータに対して高速に分散処理を行うオープンソースのフレームワークです。
大量のサイズのテキストファイルからデータをロードし、データの計算、出力する機能をサポートしていて、実際のアプリケーションでも頻繁に行われる処理です。
APIは、Scala, Java, Python などに対して提供されていますが、ここでは、Java(JUnit)を通して、テキストファイルの読み込みをテストしてみます。
また、単純にローカルのテキストを読み込むケースと、仮想HDFS環境から読み込むケースを試していきます。

#### 2.（参考）実行環境
```bash
$ java -version
openjdk version "1.8.0_252"
OpenJDK Runtime Environment (build 1.8.0_252-8u252-b09-1~18.04-b09)
OpenJDK 64-Bit Server VM (build 25.252-b09, mixed mode)
```

使用するライブラリ

|Library  |Version  |Note  |
|---|---|---|
|hadoop-minicluster  |3.3.0  |仮想HDFSクラスタを構築するため  |
|spark-core_2.11  |2.3.2  |spark API を使用するため  |
|spark-sql_2.11  |2.3.2  |spark-sql API を使用するため |
|junit  |4.12  |JUnit を使用するため  |

#### 3. 必要ライブラリの設定
手元の環境では、Maven を使用していましたので、pom.xml の一部を掲載しておきます。

```pom.xml
<dependencies>
		<!--hadoop-minicluster is first to avoid "NoSuchMethodError: com.google.common.base.Preconditions.checkArgument" -->
		<dependency>
			<groupId>org.apache.hadoop</groupId>
			<artifactId>hadoop-minicluster</artifactId>
			<version>3.3.0</version>
			<scope>test</scope>
		</dependency>
		<dependency>
			<groupId>org.apache.spark</groupId>
			<artifactId>spark-core_2.11</artifactId>
			<version>2.3.2</version>
		</dependency>
		<dependency>
			<groupId>org.apache.spark</groupId>
			<artifactId>spark-sql_2.11</artifactId>
			<version>2.3.2</version>
		</dependency>
		<dependency>
			<groupId>junit</groupId>
			<artifactId>junit</artifactId>
			<version>4.12</version>
			<scope>test</scope>
		</dependency>
	</dependencies>
```

ここで hadoop-minicluster を一番上で定義している理由は、spark のライブラリを先にロードしてしまうと、古いバージョンのライブラリが優先して適用されてしまうためです。
hadoop-minicluster のバージョンを下げるか、意図的に古いバージョンのライブラリを exclude してあげる方が正しい対処法と思いますが、簡易テストのため簡単な回避策を取りました。

#### 4. ローカルディレクトリからファイルロード（仮想HDFS環境なし）
まずは、単純にローカルディレクトリからファイルをロードするJUnitを書いてみます。

```java
import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertThat;

import java.io.BufferedWriter;
import java.io.File;
import java.io.IOException;
import java.nio.file.Paths;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import org.apache.hadoop.fs.FileUtil;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.Metadata;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.spark_project.guava.io.Files;

public class ReadLocalTextTest {
	private File tempDir;
	private String filePath;
	private SparkSession ss;
	
	@Before
	public void setUp() throws IOException {
		// Create directory
		tempDir = Files.createTempDir();
		
		// Define a text file
		filePath = tempDir.getAbsolutePath().concat("/file1.txt");
		
		// Spark configuration
		SparkConf sparkConf = new SparkConf().setAppName("ReadLocalTextTest").setMaster("local[*]")
				.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
				.set("spark.io.compression.codec", "lz4");
		ss = SparkSession.builder().config(sparkConf).getOrCreate();
	}
	
	@After
	public void cleanup() {
		if (ss != null) {
			ss.stop();
			ss = null;
		}
		FileUtil.fullyDelete(tempDir);
	}
	
	@Test
	public void readTextFile() {
		// Create a text file on local
		createTextFile();
		
		// Define the text file schema
		StructType schema = new StructType(new StructField[] {
			new StructField("id", DataTypes.IntegerType, false, Metadata.empty()),
			new StructField("message", DataTypes.StringType, false, Metadata.empty())
		});
		
		// Read the text file
		Dataset<Row> ds = ss.read()
				.option("header", false)
				.option("delimiter", "\t")
				.schema(schema).csv(filePath.toString())
				.cache();
		
		assertThat(ds.count(), is(2L));
		assertThat(ds.filter(ds.col("id").equalTo(1)).first().getAs("message"), is("This is first line."));
		assertThat(ds.filter(ds.col("id").equalTo(2)).first().getAs("message"), is("This is second line."));
	}
	
	private void createTextFile() {
		List<String> lines = Stream.of(
				"1\tThis is first line.",
				"2\tThis is second line.").collect(Collectors.toList());
		
		try (BufferedWriter writer = java.nio.file.Files.newBufferedWriter(Paths.get(filePath))) {
            for (String line : lines) {
                writer.append(line);
                writer.newLine();
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
	}
}
```

- setup()
/tmp 配下にディレクトリを作成しています。後でこちらに読み込み用のテキストファイルを作成します。
あとは、SparkSession のインスタンスを生成しています。

- createTextFile()
2行のテキストを含んだテキストファイルを作成しています。

- readTextFileTest()
spark の API を用いてファイルを読み込んでいます。
```csv(filePath.toString())``` と定義しているように、単に /tmp/{ディレクトリ}/ファイル とローカルファイルのパスを指定しているだけです。

- cleanup()
SparkSession のインスタンスを停止しています。
あとは、/tmp 配下に作成したディレクトリを削除しています。

JUnit テストを実行してみると、定義したテストケースはパスします。
また、最後に ds.show() でコンテンツを出力していますが、以下のように正しく読み込んでいることが分かります。

```
+---+--------------------+
| id|             message|
+---+--------------------+
|  1| This is first line.|
|  2|This is second line.|
+---+--------------------+
```

#### 5. 仮想HDFS環境からファイルロード
次に仮想HDFS環境を用意し、そこに置かれているファイルをロードするJUnitを書いていきます。

```java
import static org.hamcrest.CoreMatchers.is;
import static org.junit.Assert.assertThat;

import java.io.File;
import java.io.IOException;
import java.util.List;
import java.util.stream.Collectors;
import java.util.stream.Stream;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.FileUtil;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.hdfs.HdfsConfiguration;
import org.apache.hadoop.hdfs.MiniDFSCluster;
import org.apache.spark.SparkConf;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.Metadata;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import org.junit.After;
import org.junit.Before;
import org.junit.Test;
import org.spark_project.guava.io.Files;

public class ReadHDFSTextTest {
	private File tempDir;
	private String c1PathStr;
	private String filePath;

	private FileSystem fs;
	private MiniDFSCluster cluster;
	private SparkSession ss;

	@Before
	public void setup() throws Exception {
		// Create directory
		tempDir = Files.createTempDir();
		c1PathStr = tempDir.getAbsolutePath().concat("/cluster1");
		filePath = c1PathStr.concat("/").concat("dfile1.txt");

		// HDFS configuration
		Configuration conf = new HdfsConfiguration();
		conf.set(MiniDFSCluster.HDFS_MINIDFS_BASEDIR, c1PathStr);
		cluster = new MiniDFSCluster.Builder(conf).build();
		fs = FileSystem.get(conf);

		// Spark configuration
		SparkConf sparkConf = new SparkConf().setAppName("ReadHDFSTextTest").setMaster("local[*]")
				.set("spark.serializer", "org.apache.spark.serializer.KryoSerializer")
				.set("spark.io.compression.codec", "lz4");
		ss = SparkSession.builder().config(sparkConf).getOrCreate();
	}

	@After
	public void tearDown() {
		if (cluster != null) {
			cluster.shutdown();
			cluster = null;
		}
		if (ss != null) {
			ss.stop();
			ss = null;
		}
		FileUtil.fullyDelete(tempDir);
	}

	@Test
	public void readTextFileTest() throws IOException {
		// Create a text file on HDFS
		createTextFile();
		
		// Define the text file schema
		StructType schema = new StructType(
				new StructField[] { new StructField("id", DataTypes.IntegerType, false, Metadata.empty()),
						new StructField("message", DataTypes.StringType, false, Metadata.empty()) });
		
		// Read the text file
		Dataset<Row> ds = ss.read()
				.option("header", false)
				.option("delimiter", "\t")
				.schema(schema).csv("hdfs://" + fs.getCanonicalServiceName() + "/" + filePath.toString())
				.cache();
		assertThat(ds.count(), is(2L));
		assertThat(ds.filter(ds.col("id").equalTo(1)).first().getAs("message"), is("This is first line."));
		assertThat(ds.filter(ds.col("id").equalTo(2)).first().getAs("message"), is("This is second line."));
		ds.show();
	}
	
	private void createTextFile() throws IllegalArgumentException, IOException {
		FSDataOutputStream out = fs.create(new Path(filePath));
		
		// Required \n for writing with change line
		List<String> lines = Stream.of("1\tThis is first line.\n", "2\tThis is second line.")
				.collect(Collectors.toList());
		for (String line : lines) {
			out.writeBytes(line);
		}
		out.close();
	}
}
```

- setup()  
/tmp 配下にディレクトリを作成しています。また必須ではありませんが、仮想HDFS環境用のパスとして、"cluster1" というディレクトリを追加しています。後でこちらに読み込み用のテキストファイルを作成します。  
次に、HdfsConfiguration のインスタンスを作成し、MiniDFSCluster のベースディレクトリに先ほど作成した "/tmp/{ディレクトリ}/cluster1" を設定した上で、MiniDFSCluster のインスタンス(cluster)を作成しています。
また、同じ設定(HdfsConfiguration)を元に FileSystem のインスタンスを用意しておくことで、このFileSystem のインスタンスを通して、HDFSにアクセスできるようにしておきます。
あとは、SparkSession のインスタンスを生成しています。

- cleanup()
MiniDFSCluster のインスタンス(cluster)と SparkSession のインスタンスを停止しています。
あとは、/tmp 配下に作成したディレクトリを削除しています。

- createTextFile()
2行のテキストを含んだテキストファイルを作成しています。FileSystem のインスタンスから、FSDataOutputStream のインスタンスを作成し、HDFSに書き込みしています。
また、こちらのインスタンス経由では、書き込み時に自動的に改行しないため、書き込むテキスト内に意図的に改行コード(\n)を挿入しています。

- readTextFileTest()
spark の API を用いてファイルを読み込んでいます。
```csv("hdfs://" + fs.getCanonicalServiceName() + "/" + filePath.toString())``` のように、HDFS 上のファイルをロードすることを意図的に定義しています。

JUnit テストを実行してみると、定義したテストケースはパスします。
また、最後に ds.show() でコンテンツを出力していますが、以下のように正しく読み込んでいることが分かります。

```bash
+---+--------------------+
| id|             message|
+---+--------------------+
|  1| This is first line.|
|  2|This is second line.|
+---+--------------------+
```