add rtdmp into dmp project

azkaban/adn/package/adn_install_total_v2.sh

@@ -25,6 +25,10 @@ hadoop jar ../../${JAR} mobvista.dmp.main.ParseInstallRCFile \
   -Dmapreduce.fileoutputcommitter.algorithm.version=2 \
   $INPUT_PATH $OUTPUT_PATH ${REDUCE_NUM}
 
+if [ $? -ne 0 ];then
+  exit 255
+fi
+
 : '
 business="adn_install"
 

azkaban/adn/package/adn_request_other_install_v2.sh

@@ -24,6 +24,10 @@ hadoop jar ../../${JAR} mobvista.dmp.main.ParseInstallRCFile \
   -Dmapreduce.fileoutputcommitter.algorithm.version=2 \
   $INPUT_PATH $OUTPUT_PATH ${REDUCE_NUM}
 
+if [ $? -ne 0 ];then
+  exit 255
+fi
+
 : '
 business="adn_request_other"
 

azkaban/adn/package/adn_request_pkg_total_v2.sh

@@ -25,6 +25,10 @@ hadoop jar ../../${JAR} mobvista.dmp.main.ParseInstallRCFile \
   -Dmapreduce.fileoutputcommitter.algorithm.version=2 \
   $INPUT_PATH $OUTPUT_PATH ${REDUCE_NUM}
 
+if [ $? -ne 0 ];then
+  exit 255
+fi
+
 : '
 business="adn_request_sdk"
 

azkaban/rtdmp/rtdmp_alter.sh

@@ -4,8 +4,8 @@ source ../dmp_env.sh
 
 today=${ScheduleTime:-$1}
 
-start_time=$(date +"%Y-%m-%d %H:00:00" -d "-168 hours $today")
-end_time=$(date +"%Y-%m-%d %H:59:59" -d "-1 hours $today")
+start_time=$(date +"%Y-%m-%d 00:00:00" -d "-7 days $today")
+end_time=$(date +"%Y-%m-%d 23:59:59" -d "-1 days $today")
 
 java -cp ../${JAR} mobvista.dmp.datasource.rtdmp.RTDmpAlter "${start_time}" "${end_time}"
 

azkaban/rtdmp/rtdmp_fetch.job

 type=command
-command=bash -x rtdmp_fetch.sh
\ No newline at end of file
+command=bash -x rtdmp_fetch_v2.sh
\ No newline at end of file

azkaban/rtdmp/rtdmp_fetch_v2.sh

+#!/bin/bash
+
+source ../dmp_env.sh
+
+today=${ScheduleTime:-$1}
+
+start_time=$(date +"%Y-%m-%d 00:00:00" -d "-7 days $today")
+end_time=$(date +"%Y-%m-%d 23:59:59" -d "-1 days $today")
+
+if [[ ! -d "/home/hadoop/wangjf/DmpServer/data" ]]; then
+  mkdir -p /home/hadoop/wangjf/DmpServer/data
+fi
+
+java -cp ../${JAR} mobvista.dmp.datasource.rtdmp.RTDmpFetch "${start_time}" "${end_time}"
+
+if [[ $? -ne 0 ]]; then
+    exit 255
+fi
\ No newline at end of file

pom.xml

@@ -288,6 +288,11 @@
             <artifactId>alipay-sdk-java</artifactId>
             <version>4.10.192.ALL</version>
         </dependency>
+        <dependency>
+            <groupId>com.amazonaws</groupId>
+            <artifactId>aws-java-sdk-s3</artifactId>
+            <version>1.11.437</version>
+        </dependency>
     </dependencies>
 
     <build>

src/main/java/mobvista/dmp/datasource/rtdmp/entity/Tuple.java

+package mobvista.dmp.datasource.rtdmp.entity;
+
+public class Tuple {
+
+    private Boolean flag;
+
+    private Long utime;
+
+    public Boolean getFlag() {
+        return flag;
+    }
+
+    public void setFlag(Boolean flag) {
+        this.flag = flag;
+    }
+
+    public Long getUtime() {
+        return utime;
+    }
+
+    public void setUtime(Long utime) {
+        this.utime = utime;
+    }
+
+    public Tuple(Boolean flag, Long utime) {
+        this.flag = flag;
+        this.utime = utime;
+    }
+}

src/main/java/mobvista/dmp/util/AwsUploadFileToS3.java

+package mobvista.dmp.util;
+
+import com.amazonaws.AmazonServiceException;
+import com.amazonaws.SdkClientException;
+import com.amazonaws.auth.AWSCredentials;
+import com.amazonaws.auth.AWSStaticCredentialsProvider;
+import com.amazonaws.auth.BasicAWSCredentials;
+import com.amazonaws.regions.Regions;
+import com.amazonaws.services.s3.AmazonS3;
+import com.amazonaws.services.s3.AmazonS3ClientBuilder;
+import com.amazonaws.services.s3.transfer.MultipleFileUpload;
+import com.amazonaws.services.s3.transfer.TransferManager;
+import com.amazonaws.services.s3.transfer.TransferManagerBuilder;
+
+import java.io.File;
+import java.util.ArrayList;
+import java.util.List;
+
+/**
+ * @package: mobvista.dmp.util
+ * @author: wangjf
+ * @date: 2021/8/3
+ * @time: 11:22 上午
+ * @email: jinfeng.wang@mobvista.com
+ */
+public class AwsUploadFileToS3 {
+    private static Regions clientRegion = Regions.US_EAST_1;
+    private static AWSCredentials credentials = new BasicAWSCredentials(
+            PropertyUtil.getProperty("config.properties", "aws.accessKey"),
+            PropertyUtil.getProperty("config.properties", "aws.secretKey")
+    );
+
+    private static AmazonS3 s3Client = AmazonS3ClientBuilder.standard()
+            .withRegion(clientRegion)
+            .withCredentials(new AWSStaticCredentialsProvider(credentials))
+            .build();
+
+    public void uploadFileList(List<String> file_paths, Integer jobId, String deviceType, String part) {
+        String bucketName = PropertyUtil.getProperty("config.properties", "aws.s3.bucket");
+        String keyName = PropertyUtil.getProperty("config.properties", "aws.s3.key") + "/" + jobId + "/" + part + "/" + deviceType;
+
+        ArrayList<File> files = new ArrayList<File>();
+        for (String path : file_paths) {
+            files.add(new File(path));
+        }
+
+        TransferManager xfer_mgr = TransferManagerBuilder.standard()
+                .withS3Client(s3Client)
+                .build();
+        try {
+            MultipleFileUpload xfer = xfer_mgr.uploadFileList(bucketName,
+                    keyName, new File(PropertyUtil.getProperty("config.properties", "rtdmp.output.dir")), files);
+            XferMgrProgress.showTransferProgress(xfer);
+            XferMgrProgress.waitForCompletion(xfer);
+        } catch (AmazonServiceException e) {
+            System.err.println(e.getErrorMessage());
+            System.exit(1);
+        }
+        xfer_mgr.shutdownNow(false);
+        for (String path : file_paths) {
+            File deleteFile = new File(path);
+            //  上传完完后即删除原文件
+            if (deleteFile.exists()) {
+                deleteFile.delete();
+            }
+        }
+    }
+}

src/main/java/mobvista/dmp/util/ZipFileUtil.java

+package mobvista.dmp.util;
+
+import org.slf4j.Logger;
+import org.slf4j.LoggerFactory;
+
+import java.io.*;
+import java.util.zip.GZIPOutputStream;
+import java.util.zip.ZipEntry;
+import java.util.zip.ZipOutputStream;
+
+/**
+ * @package: mobvista.dmp.util
+ * @author: wangjf
+ * @date: 2021/8/3
+ * @time: 11:21 上午
+ * @email: jinfeng.wang@mobvista.com
+ */
+public class ZipFileUtil {
+
+    public static final Logger logger = LoggerFactory.getLogger(ZipFileUtil.class);
+
+    /**
+     * 压缩单个文件
+     */
+    public static void zipFile(String filePath, String zipPath) {
+        try {
+            File file = new File(filePath);
+            File zipFile = new File(zipPath);
+            InputStream input = new FileInputStream(file);
+            ZipOutputStream zipOut = new ZipOutputStream(new FileOutputStream(zipFile));
+            zipOut.putNextEntry(new ZipEntry(file.getName()));
+            int temp = 0;
+            while ((temp = input.read()) != -1) {
+                zipOut.write(temp);
+            }
+            input.close();
+            zipOut.close();
+        } catch (Exception e) {
+            e.printStackTrace();
+        }
+    }
+
+    public static void compress(String filePath, String gzipPath) {
+        byte[] buffer = new byte[1024];
+        try {
+            //  Specify Name and Path of Output GZip file here
+            GZIPOutputStream gos = new GZIPOutputStream(new FileOutputStream(gzipPath));
+            //  Specify the input file here
+            FileInputStream fis = new FileInputStream(filePath);
+            //  Read from input file and write to output GZip file
+            int length;
+            //  fis.read(buffer)， 结果时Buffer有了内容，同时返回读取内容的长度，读到文件末尾时读取内容的长度变为-1
+            while ((length = fis.read(buffer)) > 0) {
+                gos.write(buffer, 0, length);
+            }
+            fis.close();
+            gos.finish();
+            gos.close();
+            System.out.println(filePath + ",File Compressed!");
+        } catch (IOException e) {
+            e.printStackTrace();
+        }
+    }
+}

src/main/resources/config.properties

@@ -77,6 +77,13 @@ rtdmp.noauto.business=dsp_req,btop
 
 rtdmp.auto.business=ali_activation
 
+aws.s3.bucket=mob-emr-test
+aws.s3.key=dataplatform/fmp/data/rtdmp
+aws.accessKey=AKIA35FGARBHBMBRCBHG
+aws.secretKey=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
+
+rtdmp.output.dir=/home/hadoop/wangjf/DmpServer/data/
+
 # alipay_activation,alipay_acquisition,uc_activation,iqiyi_activation,youku_acquisition
 
 ali_activation.package_name=com.taobao.foractivation.172393,com.taobao.foractivation.184287,com.taobao.foractivation.184289,\

src/main/resources/logback.xml

@@ -21,7 +21,7 @@
             </Pattern>
         </layout>
     </appender>
-    <root level="warn">
+    <root level="info">
         <appender-ref ref="console"/>
     </root>