如何在Apache Spark上运行Apache Tika 集群模式文本提取为空如何解决

问题根因 #

你遇到的集群模式返回空文本是Apache Tika的SPI服务加载机制导致的：

• Tika依赖Java SPI（服务提供者接口）动态加载不同格式文档的解析器实现，相关配置存放在JAR包的META-INF/services/org.apache.tika.parser.Parser文件中
• 构建shaded JAR时如果没有特殊配置，多个依赖JAR里的同名SPI配置文件会互相覆盖，最终生成的JAR里缺少完整的解析器注册配置
• local模式下运行时会加载本地环境中完整的Tika依赖配置，所以运行正常，集群Executor节点只能加载到你提交的不完整的shaded JAR，AutoDetectParser找不到对应格式的解析器，就会返回空字符串

解决方案 #

1. 修正shaded JAR打包配置 #

构建Fat JAR时必须合并所有SPI服务配置文件，不要直接覆盖：

Maven配置示例（maven-shade-plugin）

在pom.xml的shade插件配置中添加ServicesResourceTransformer：

<plugin>
  <groupId>org.apache.maven.plugins</groupId>
  <artifactId>maven-shade-plugin</artifactId>
  <version>3.2.4</version>
  <executions>
    <execution>
      <phase>package</phase>
      <goals>
        <goal>shade</goal>
      </goals>
      <configuration>
        <transformers>
          <transformer implementation="org.apache.maven.plugins.shade.resource.ServicesResourceTransformer"/>
          <!-- 保留Tika的其他配置文件 -->
          <transformer implementation="org.apache.maven.plugins.shade.resource.IncludeResourceTransformer">
            <resource>META-INF/services/org.apache.tika.parser.Parser</resource>
          </transformer>
        </transformers>
      </configuration>
    </execution>
  </executions>
</plugin>

SBT配置示例（sbt-assembly）

在build.sbt中添加合并策略：

assemblyMergeStrategy in assembly := {
  case PathList("META-INF", "services", _*) => MergeStrategy.concat
  case x =>
    val oldStrategy = (assemblyMergeStrategy in assembly).value
    oldStrategy(x)
}

2. 调整Spark提交参数 #

提交任务时配置用户类路径优先，避免和EMR集群自带的旧版本Tika依赖冲突：

spark-submit \
  --class TikaTest \
  --master yarn \
  --deploy-mode cluster \
  --conf spark.driver.userClassPathFirst=true \
  --conf spark.executor.userClassPathFirst=true \
  your-tika-fatjar.jar \
  s3://输入路径 \
  s3://输出路径

3. 优化代码添加异常捕获 #

原代码没有异常捕获逻辑，解析出错也会静默返回空，添加异常处理方便定位问题，同时取消BodyContentHandler默认的字符数限制，避免长文档被截断：

def parseFile(stream: DataInputStream): String = {
  val parser = new AutoDetectParser()
  val handler = new BodyContentHandler(-1) 
  val metadata = new Metadata()
  val context = new ParseContext()
  try {
    parser.parse(stream, handler, metadata, context)
    handler.toString()
  } catch {
    case e: Exception => 
      s"[解析失败] ${e.getMessage}"
  } finally {
    stream.close()
  }
}

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