[hotfix-DTStack#1821][hdfs]Modify orc file calculation method

ll076110 · Sep 21, 2023 · ace6540 · ace6540
1 parent 45adcc0
commit ace6540
Show file tree

Hide file tree

Showing 5 changed files with 69 additions and 60 deletions.
diff --git a/...ector-hdfs/src/main/java/com/dtstack/chunjun/connector/hdfs/sink/HdfsOrcOutputFormat.java b/...ector-hdfs/src/main/java/com/dtstack/chunjun/connector/hdfs/sink/HdfsOrcOutputFormat.java
@@ -149,7 +149,7 @@ protected void openSource() {
     @Override
     // todo the deviation needs to be calculated accurately
     protected long getCurrentFileSize() {
-        return (long) (bytesWriteCounter.getLocalValue() * getDeviation());
+        return (long) ((bytesWriteCounter.getLocalValue() - lastWriteSize) * getDeviation());
     }
 
     @Override

diff --git a/...r-hdfs/src/main/java/com/dtstack/chunjun/connector/hdfs/sink/HdfsParquetOutputFormat.java b/...r-hdfs/src/main/java/com/dtstack/chunjun/connector/hdfs/sink/HdfsParquetOutputFormat.java
@@ -156,7 +156,7 @@ protected void nextBlock() {
     public void flushDataInternal() {
         log.info(
                 "Close current parquet record writer, write data size:[{}]",
-                SizeUnitType.readableFileSize(bytesWriteCounter.getLocalValue()));
+                SizeUnitType.readableFileSize(bytesWriteCounter.getLocalValue() - lastWriteSize));
         try {
             if (writer != null) {
                 writer.close();

diff --git a/...ctor-hdfs/src/main/java/com/dtstack/chunjun/connector/hdfs/sink/HdfsTextOutputFormat.java b/...ctor-hdfs/src/main/java/com/dtstack/chunjun/connector/hdfs/sink/HdfsTextOutputFormat.java
@@ -77,7 +77,7 @@ protected void nextBlock() {
     public void flushDataInternal() {
         log.info(
                 "Close current text stream, write data size:[{}]",
-                SizeUnitType.readableFileSize(bytesWriteCounter.getLocalValue()));
+                SizeUnitType.readableFileSize(bytesWriteCounter.getLocalValue() - lastWriteSize));
 
         try {
             if (stream != null) {

diff --git a/chunjun-core/src/main/java/com/dtstack/chunjun/sink/format/BaseFileOutputFormat.java b/chunjun-core/src/main/java/com/dtstack/chunjun/sink/format/BaseFileOutputFormat.java
@@ -58,6 +58,7 @@ public abstract class BaseFileOutputFormat extends BaseRichOutputFormat {
     protected List<String> preCommitFilePathList = new ArrayList<>();
     protected long nextNumForCheckDataSize;
     protected long lastWriteTime = System.currentTimeMillis();
+    protected long lastWriteSize;
 
     @Override
     public void initializeGlobal(int parallelism) {
@@ -150,6 +151,7 @@ public void flushData() {
         if (rowsOfCurrentBlock != 0) {
             flushDataInternal();
             sumRowsOfBlock += rowsOfCurrentBlock;
+            lastWriteSize = bytesWriteCounter.getLocalValue();
             log.info(
                     "flush file:{}, rowsOfCurrentBlock = {}, sumRowsOfBlock = {}",
                     currentFileName,

diff --git a/chunjun-local-test/pom.xml b/chunjun-local-test/pom.xml
@@ -51,71 +51,78 @@
 			<artifactId>chunjun-connector-stream</artifactId>
 			<version>${project.version}</version>
 		</dependency>
-		<dependency>
-			<groupId>com.dtstack.chunjun</groupId>
-			<artifactId>chunjun-connector-binlog</artifactId>
-			<version>${project.version}</version>
-		</dependency>
+		<!--rdb test start-->
+<!--		<dependency>-->
+<!--			<groupId>com.dtstack.chunjun</groupId>-->
+<!--			<artifactId>chunjun-connector-mysql</artifactId>-->
+<!--			<version>${project.version}</version>-->
+<!--		</dependency>-->
+		<!--rdb test end-->
+
+		<!--hdfs test start-->
+<!--		<dependency>-->
+<!--			<groupId>com.dtstack.chunjun</groupId>-->
+<!--			<artifactId>chunjun-connector-hdfs</artifactId>-->
+<!--			<version>${project.version}</version>-->
+<!--		</dependency>-->
+<!--		<dependency>-->
+<!--			<artifactId>hadoop-mapreduce-client-core</artifactId>-->
+<!--			<exclusions>-->
+<!--				<exclusion>-->
+<!--					<artifactId>commons-cli</artifactId>-->
+<!--					<groupId>commons-cli</groupId>-->
+<!--				</exclusion>-->
+<!--				<exclusion>-->
+<!--					<artifactId>snappy-java</artifactId>-->
+<!--					<groupId>org.xerial.snappy</groupId>-->
+<!--				</exclusion>-->
+<!--			</exclusions>-->
+<!--			<groupId>org.apache.hadoop</groupId>-->
+<!--			<version>2.7.5</version>-->
+<!--		</dependency>-->
+<!--		<dependency>-->
+<!--			<artifactId>hadoop-common</artifactId>-->
+<!--			<exclusions>-->
+<!--				<exclusion>-->
+<!--					<artifactId>gson</artifactId>-->
+<!--					<groupId>com.google.code.gson</groupId>-->
+<!--				</exclusion>-->
+<!--				<exclusion>-->
+<!--					<artifactId>commons-cli</artifactId>-->
+<!--					<groupId>commons-cli</groupId>-->
+<!--				</exclusion>-->
+<!--				<exclusion>-->
+<!--					<artifactId>snappy-java</artifactId>-->
+<!--					<groupId>org.xerial.snappy</groupId>-->
+<!--				</exclusion>-->
+<!--			</exclusions>-->
+<!--			<groupId>org.apache.hadoop</groupId>-->
+<!--			<version>2.7.5</version>-->
+<!--		</dependency>-->
+<!--		<dependency>-->
+<!--			<artifactId>hadoop-hdfs</artifactId>-->
+<!--			<exclusions>-->
+<!--				<exclusion>-->
+<!--					<artifactId>commons-cli</artifactId>-->
+<!--					<groupId>commons-cli</groupId>-->
+<!--				</exclusion>-->
+<!--				<exclusion>-->
+<!--					<artifactId>netty-all</artifactId>-->
+<!--					<groupId>io.netty</groupId>-->
+<!--				</exclusion>-->
+<!--			</exclusions>-->
+<!--			<groupId>org.apache.hadoop</groupId>-->
+<!--			<version>2.7.5</version>-->
+<!--		</dependency>-->
+		<!--hdfs test end-->
 
-		<dependency>
-			<groupId>com.dtstack.chunjun</groupId>
-			<artifactId>chunjun-connector-iceberg</artifactId>
-			<version>${project.version}</version>
-		</dependency>
-		<dependency>
-			<groupId>com.dtstack.chunjun</groupId>
-			<artifactId>chunjun-connector-mongodb</artifactId>
-			<version>${project.version}</version>
-		</dependency>
-
-		<dependency>
-			<groupId>com.dtstack.chunjun</groupId>
-			<artifactId>chunjun-connector-emqx</artifactId>
-			<version>${project.version}</version>
-		</dependency>
-		<dependency>
-			<groupId>com.dtstack.chunjun</groupId>
-			<artifactId>chunjun-connector-jdbc-base</artifactId>
-			<version>${project.version}</version>
-		</dependency>
-
-		<dependency>
-			<groupId>com.dtstack.chunjun</groupId>
-			<artifactId>chunjun-connector-mysql</artifactId>
-			<version>${project.version}</version>
-		</dependency>
-
-		<dependency>
-			<groupId>com.dtstack.chunjun</groupId>
-			<artifactId>chunjun-connector-kafka</artifactId>
-			<version>${project.version}</version>
-		</dependency>
-
-		<dependency>
-			<groupId>com.dtstack.chunjun</groupId>
-			<artifactId>chunjun-connector-kingbase</artifactId>
-			<version>${project.version}</version>
-		</dependency>
-
-		<!-- -->
-		<dependency>
-			<groupId>com.dtstack.chunjun</groupId>
-			<artifactId>chunjun-connector-ftp</artifactId>
-			<version>${project.version}</version>
-		</dependency>
 
 		<dependency>
 			<groupId>org.apache.flink</groupId>
 			<artifactId>flink-json</artifactId>
 			<version>${flink.version}</version>
 		</dependency>
 
-		<dependency>
-			<groupId>com.dtstack.chunjun</groupId>
-			<artifactId>chunjun-connector-oceanbase</artifactId>
-			<version>master</version>
-		</dependency>
-
 		<dependency>
 			<groupId>org.apache.avro</groupId>
 			<artifactId>avro</artifactId>