Adapt snappy-raw compression read (#119)

* adapt afs * comment afs dependency Change-Id: Ie38ba73573d12edcfd7eaac22a3b041c781dc7f6
apache · Nov 4, 2020 · 4802fec · 4802fec
1 parent b6f065b
commit 4802fec
Show file tree

Hide file tree

Showing 4 changed files with 32 additions and 6 deletions.
diff --git a/assembly/static/bin/hugegraph-loader.sh b/assembly/static/bin/hugegraph-loader.sh
@@ -14,6 +14,7 @@ BIN=`abs_path`
 TOP="$(cd ${BIN}/../ && pwd)"
 CONF="$TOP/conf"
 LIB="$TOP/lib"
+NATIVE="$TOP/native"
 LOG="$TOP/logs"
 
 # Use the unofficial bash strict mode to avoid subtle bugs impossible.
@@ -61,4 +62,5 @@ export JVM_OPTS="$JVM_OPTS -Xmx10g -cp $LOADER_CLASSPATH"
 #JVM_OPTS="$JVM_OPTS -agentlib:jdwp=transport=dt_socket,server=y,suspend=y,address=1414"
 
 exec "$JAVA" -Dname="HugeGraphLoader" -Dlog4j.configurationFile="$CONF/log4j2.xml" \
+-Djava.library.path=${NATIVE} \
 ${JVM_OPTS} com.baidu.hugegraph.loader.HugeGraphLoader ${VARS}
diff --git a/pom.xml b/pom.xml
@@ -154,6 +154,13 @@
             <artifactId>hadoop-hdfs-client</artifactId>
             <version>2.8.5</version>
         </dependency>
+        <!--
+        <dependency>
+            <groupId>com.baidu.inf.hadoop</groupId>
+            <artifactId>libdfs-java</artifactId>
+            <version>2.0.5</version>
+        </dependency>
+        -->
         <dependency>
             <groupId>org.apache.hadoop</groupId>
             <artifactId>hadoop-mapred</artifactId>
@@ -172,6 +179,11 @@
             <artifactId>commons-compress</artifactId>
             <version>1.19</version>
         </dependency>
+        <dependency>
+            <groupId>org.apache.httpcomponents</groupId>
+            <artifactId>httpclient</artifactId>
+            <version>4.5.10</version>
+        </dependency>
         <dependency>
             <groupId>org.tukaani</groupId>
             <artifactId>xz</artifactId>
@@ -210,6 +222,10 @@
                     <groupId>org.slf4j</groupId>
                     <artifactId>slf4j-api</artifactId>
                 </exclusion>
+                <exclusion>
+                    <groupId>org.slf4j</groupId>
+                    <artifactId>slf4j-log4j12</artifactId>
+                </exclusion>
                 <exclusion>
                     <groupId>org.apache.logging.log4j</groupId>
                     <artifactId>log4j-slf4j-impl</artifactId>
@@ -275,8 +291,8 @@
                     <artifactId>parquet-column</artifactId>
                 </exclusion>
                 <exclusion>
-                    <artifactId>libfb303</artifactId>
                     <groupId>org.apache.thrift</groupId>
+                    <artifactId>libfb303</artifactId>
                 </exclusion>
             </exclusions>
         </dependency>

diff --git a/src/main/java/com/baidu/hugegraph/loader/reader/file/FileLineFetcher.java b/src/main/java/com/baidu/hugegraph/loader/reader/file/FileLineFetcher.java
@@ -33,6 +33,11 @@
 import org.apache.commons.compress.compressors.CompressorInputStream;
 import org.apache.commons.compress.compressors.CompressorStreamFactory;
 import org.apache.commons.lang3.StringUtils;
+import org.apache.hadoop.conf.Configuration;
+import org.apache.hadoop.io.compress.CompressionCodec;
+import org.apache.hadoop.io.compress.CompressionInputStream;
+import org.apache.hadoop.io.compress.SnappyCodec;
+import org.apache.hadoop.util.ReflectionUtils;
 import org.slf4j.Logger;
 
 import com.baidu.hugegraph.loader.exception.LoadException;
@@ -252,11 +257,17 @@ private static Reader createCompressReader(InputStream stream,
         switch (compression) {
             case NONE:
                 return new InputStreamReader(stream, charset);
+            case SNAPPY_RAW:
+                Configuration config = new Configuration();
+                CompressionCodec codec = ReflectionUtils.newInstance(
+                                         SnappyCodec.class, config);
+                CompressionInputStream sis = codec.createInputStream(
+                                             stream, codec.createDecompressor());
+                return new InputStreamReader(sis, charset);
             case GZIP:
             case BZ2:
             case XZ:
             case LZMA:
-            case SNAPPY_RAW:
             case SNAPPY_FRAMED:
             case Z:
             case DEFLATE:

diff --git a/src/test/java/com/baidu/hugegraph/loader/test/functional/FileLoadTest.java b/src/test/java/com/baidu/hugegraph/loader/test/functional/FileLoadTest.java
@@ -1683,10 +1683,7 @@ public void testLZMACompressFile() {
         Assert.assertEquals(1, vertices.size());
     }
 
-    /**
-     * Didn't find a way to generate the compression file using code
-     */
-    //@Test
+//    @Test
     public void testSnappyRawCompressFile() {
         ioUtil.write("vertex_person.snappy", Compression.SNAPPY_RAW,
                      "name,age,city",