Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

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

Open
wants to merge 3 commits into
base: master
Choose a base branch
from
Open

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

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
3 commits
Select commit Hold shift + click to select a range
ace6540
[hotfix-#1821][hdfs]Modify orc file calculation method
Sep 21, 2023
70c00af
Merge branch 'master' into hotfix_1821
ll076110 Sep 21, 2023
9b075a0
Merge branch 'master' into hotfix_1821
ll076110 Oct 18, 2023
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion ...ector-hdfs/src/main/java/com/dtstack/chunjun/connector/hdfs/sink/HdfsOrcOutputFormat.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down
2 changes: 1 addition & 1 deletion ...r-hdfs/src/main/java/com/dtstack/chunjun/connector/hdfs/sink/HdfsParquetOutputFormat.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down
2 changes: 1 addition & 1 deletion ...ctor-hdfs/src/main/java/com/dtstack/chunjun/connector/hdfs/sink/HdfsTextOutputFormat.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down
2 changes: 2 additions & 0 deletions chunjun-core/src/main/java/com/dtstack/chunjun/sink/format/BaseFileOutputFormat.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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) {
Expand Down Expand Up @@ -150,6 +151,7 @@ public void flushData() {
if (rowsOfCurrentBlock != 0) {
flushDataInternal();
sumRowsOfBlock += rowsOfCurrentBlock;
lastWriteSize = bytesWriteCounter.getLocalValue();
log.info(
"flush file:{}, rowsOfCurrentBlock = {}, sumRowsOfBlock = {}",
currentFileName,
Expand Down
121 changes: 64 additions & 57 deletions chunjun-local-test/pom.xml
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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>
Expand Down