自定义Dataflow Classic Java模板编译失败：tar.gz扩展改造求助

一、标准Maven项目目录结构 #

严格遵循以下结构，确保Maven能正确识别源码和资源：

your-project-root/
├── pom.xml
└── src/
    └── main/
        ├── java/
        │   └── com/
        │       └── yourdomain/
        │           └── yourproject/
        │               ├── BulkDecompressTarGz.java  # 改造后的主类
        │               └── TarExtractDoFn.java      # 自定义tar提取DoFn
        └── resources/
            └── META-INF/
                └── services/
                    └── org.apache.beam.sdk.PipelineOptionsRegistrar  # 可选，按需添加

二、核心pom.xml配置 #

以下是适配GCP Beam模板、支持tar.gz处理的关键配置，替换占位符为你的项目信息：

<project xmlns="http://maven.apache.org/POM/4.0.0"
         xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
    <modelVersion>4.0.0</modelVersion>

    <groupId>com.yourdomain</groupId>
    <artifactId>bulk-decompress-targz</artifactId>
    <version>1.0.0</version>
    <packaging>jar</packaging>

    <properties>
        <beam.version>2.54.0</beam.version> <!-- 匹配GCP模板使用的Beam版本 -->
        <java.version>11</java.version>
        <maven.compiler.source>${java.version}</maven.compiler.source>
        <maven.compiler.target>${java.version}</maven.compiler.target>
        <project.build.sourceEncoding>UTF-8</project.build.sourceEncoding>
    </properties>

    <dependencies>
        <!-- Apache Beam核心依赖 -->
        <dependency>
            <groupId>org.apache.beam</groupId>
            <artifactId>beam-sdks-java-core</artifactId>
            <version>${beam.version}</version>
        </dependency>
        <!-- GCP Cloud Storage依赖 -->
        <dependency>
            <groupId>org.apache.beam</groupId>
            <artifactId>beam-sdks-java-io-google-cloud-platform</artifactId>
            <version>${beam.version}</version>
        </dependency>
        <!-- 处理tar.gz的依赖 -->
        <dependency>
            <groupId>org.apache.commons</groupId>
            <artifactId>commons-compress</artifactId>
            <version>1.24.0</version>
        </dependency>
        <!-- 日志依赖，按需添加 -->
        <dependency>
            <groupId>org.slf4j</groupId>
            <artifactId>slf4j-api</artifactId>
            <version>1.7.36</version>
        </dependency>
    </dependencies>

    <build>
        <plugins>
            <!-- 打包成可运行的胖Jar，适配Beam作业 -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-shade-plugin</artifactId>
                <version>3.4.1</version>
                <executions>
                    <execution>
                        <phase>package</phase>
                        <goals>
                            <goal>shade</goal>
                        </goals>
                        <configuration>
                            <createDependencyReducedPom>false</createDependencyReducedPom>
                            <transformers>
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
                                <transformer implementation="org.apache.maven.plugins.shade.resource.ManifestResourceTransformer">
                                    <mainClass>com.yourdomain.yourproject.BulkDecompressTarGz</mainClass>
                                </transformer>
                            </transformers>
                        </configuration>
                    </execution>
                </executions>
            </plugin>
            <!-- 编译插件确保Java版本匹配 -->
            <plugin>
                <groupId>org.apache.maven.plugins</groupId>
                <artifactId>maven-compiler-plugin</artifactId>
                <version>3.11.0</version>
                <configuration>
                    <source>${java.version}</source>
                    <target>${java.version}</target>
                </configuration>
            </plugin>
        </plugins>
    </build>
</project>

三、编译与打包命令 #

在项目根目录执行以下命令，跳过测试加快编译：

mvn clean package -DskipTests

编译成功后，目标Jar包会生成在target/目录下，命名为bulk-decompress-targz-1.0.0.jar（对应你的artifactId和version）。

四、tar.gz提取逻辑示例 #

在自定义DoFn中添加tar文件提取逻辑，示例代码片段：

import org.apache.commons.compress.archivers.tar.TarArchiveEntry;
import org.apache.commons.compress.archivers.tar.TarArchiveInputStream;
import org.apache.commons.compress.compressors.gzip.GzipCompressorInputStream;
import org.apache.beam.sdk.transforms.DoFn;
import org.apache.beam.sdk.values.KV;

import java.io.IOException;
import java.io.InputStream;

public class TarExtractDoFn extends DoFn<KV<String, InputStream>, KV<String, InputStream>> {

    @ProcessElement
    public void processElement(ProcessContext c) throws IOException {
        String inputGcsPath = c.element().getKey();
        InputStream gzInputStream = c.element().getValue();
        
        // 先解压gzip，再处理tar
        try (GzipCompressorInputStream gzipIn = new GzipCompressorInputStream(gzInputStream);
             TarArchiveInputStream tarIn = new TarArchiveInputStream(gzipIn)) {
            
            TarArchiveEntry entry;
            while ((entry = tarIn.getNextTarEntry()) != null) {
                if (!entry.isDirectory()) {
                    // 构造输出路径，例如原路径去掉.tar.gz后缀后加上文件名
                    String outputPath = inputGcsPath.replace(".tar.gz", "") + "/" + entry.getName();
                    // 将每个tar条目作为独立的InputStream输出
                    c.output(KV.of(outputPath, tarIn));
                }
            }
        }
    }
}

内容的提问来源于stack exchange，提问作者kbenzakrilelp