add lzo conpress for hdfs wtire with text type

Author: nanxiuzi

flinkx-hdfs/flinkx-hdfs-core/src/main/java/com/dtstack/flinkx/hdfs/ECompressType.java

@@ -32,6 +32,7 @@ public enum ECompressType {
      */
     TEXT_GZIP("GZIP", "text", ".gz", 0.331F),
     TEXT_BZIP2("BZIP2", "text", ".bz2", 0.259F),
+    TEXT_LZO("LZO", "text", ".lzo", 1.0F),
     TEXT_NONE("NONE", "text", "", 0.637F),
 
     /**

flinkx-hdfs/flinkx-hdfs-writer/pom.xml

@@ -34,6 +34,24 @@ under the License.
     </properties>
 
     <dependencies>
+        <!--lzop-->
+        <dependency>
+            <groupId>org.anarres.lzo</groupId>
+            <artifactId>lzo-core</artifactId>
+            <version>1.0.2</version>
+        </dependency>
+        <dependency>
+            <groupId>org.anarres.lzo</groupId>
+            <artifactId>lzo-hadoop</artifactId>
+            <version>1.0.5</version>
+            <exclusions>
+                <exclusion>
+                    <artifactId>hadoop-core</artifactId>
+                    <groupId>org.apache.hadoop</groupId>
+                </exclusion>
+            </exclusions>
+        </dependency>
+        <!--lzop-->
         <dependency>
             <groupId>com.dtstack.flinkx</groupId>
             <artifactId>flinkx-hdfs-core</artifactId>

flinkx-hdfs/flinkx-hdfs-writer/src/main/java/com/dtstack/flinkx/hdfs/writer/HdfsTextOutputFormat.java

@@ -26,8 +26,11 @@
 import org.apache.commons.compress.compressors.bzip2.BZip2CompressorOutputStream;
 import org.apache.commons.compress.compressors.gzip.GzipCompressorOutputStream;
 import org.apache.flink.types.Row;
+import org.apache.hadoop.conf.Configuration;
 import org.apache.hadoop.fs.Path;
 import org.apache.hadoop.hive.common.type.HiveDecimal;
+import org.apache.hadoop.io.compress.CompressionCodecFactory;
+
 import java.io.IOException;
 import java.io.OutputStream;
 import java.math.BigDecimal;
@@ -93,6 +96,9 @@ protected void nextBlock(){
                     stream = new GzipCompressorOutputStream(fs.create(p));
                 } else if(compressType == ECompressType.TEXT_BZIP2){
                     stream = new BZip2CompressorOutputStream(fs.create(p));
+                } else if (compressType == ECompressType.TEXT_LZO) {
+                    CompressionCodecFactory factory = new CompressionCodecFactory(new Configuration());
+                    stream = factory.getCodecByClassName("com.hadoop.compression.lzo.LzopCodec").createOutputStream(fs.create(p));
                 }
             }