如何使用wholeTextFiles在Spark中读取gz文件

亚尼夫·多嫩菲尔德

我有一个包含许多小.gz文件(压缩的csv文本文件)的文件夹。我需要在我的Spark作业中阅读它们,但是事情是我需要根据文件名中的信息进行一些处理。因此,我没有使用:

JavaRDD<<String>String> input = sc.textFile(...)

因为据我了解,我无法通过这种方式访问​​文件名。相反,我使用了:

JavaPairRDD<<String>String,String> files_and_content = sc.wholeTextFiles(...);

因为这样我得到了一对文件名和内容。但是,似乎这种方式使输入阅读器无法从gz文件中读取文本,而只能读取二进制的“乱码”。

因此,我想知道是否可以将其设置为以某种方式读取文本,或者使用以下方式访问文件名: sc.textFile(...)

亚伦曼

您无法读取具有WholeTextFiles的gzip压缩文件,因为它使用CombineFileInputFormat,后者由于无法拆分而无法读取gzip压缩文件(源证明):

  override def createRecordReader(
      split: InputSplit,
      context: TaskAttemptContext): RecordReader[String, String] = {

    new CombineFileRecordReader[String, String](
      split.asInstanceOf[CombineFileSplit],
      context,
      classOf[WholeTextFileRecordReader])
  }

您可以使用newAPIHadoopFilewholefileinputformat(没有内置的Hadoop,但在互联网上),以正确地得到这个工作。

更新1:我认为WholeFileInputFormat不起作用,因为它仅获取文件的字节,这意味着您可能必须编写自己的类,可能扩展WholeFileInputFormat以确保解压缩字节。

另一种选择是使用GZipInputStream自己解压缩字节

更新2:如果您可以像下面的OP注释中那样访问目录名,则可以获取所有这样的文件。

Path path = new Path("");
FileSystem fileSystem = path.getFileSystem(new Configuration()); //just uses the default one
FileStatus []  fileStatuses = fileSystem.listStatus(path);
ArrayList<Path> paths = new ArrayList<>();
for (FileStatus fileStatus : fileStatuses) paths.add(fileStatus.getPath());

本文收集自互联网,转载请注明来源。

如有侵权,请联系 [email protected] 删除。

编辑于
0

我来说两句

0 条评论
登录 后参与评论

相关文章

TOP 榜单

  1. 1

    来自Microsoft Office加载项taskpane.js的MySQL驱动程序模块的空引用

  2. 2

    使用AWS Cognito和React的仅限Facebook / Google的登录名(无用户名/密码)

  3. 3

    创建Windows Phone 8应用并将其连接到数据库的最佳方法(最好是SQL Server)

  4. 4

    为什么Java中的System.out.println()打印到控制台?

  5. 5

    卷曲函数无法解析来自bash中变量的代理

  6. 6

    是什么在Android的consumer-rules.pro和proguard-rules.pro之间的区别?

  7. 7

    设置与Apache POI Excel表散点图标记图标的颜色

  8. 8

    将Qt Pyside2与asyncio await语法一起使用?

  9. 9

    崇高的文字+蟒蛇的蟒蛇

  10. 10

    任务':app:minifyReleaseWithR8'.java.lang.NullPointerException的执行失败(无错误消息)

  11. 11

    OpenJDK的和AdoptOpenJDK的区别

  12. 12

    大型数据集缓存到Spark内存中时,“超出了GC开销限制”(通过sparklyr和RStudio)

  13. 13

    “执行测试CMAKE_HAVE_LIBC_PTHREAD”失败实际上是什么意思?

  14. 14

    使用Core 2.2中的Identity,如何在关闭浏览器15分钟后保持会话活动?

  15. 15

    React中的ForwardRefExoticComponent和ForwardRefRenderFunction有什么区别?

  16. 16

    猫鼬查找结果,然后将字段替换为findOne

  17. 17

    如何降级Google Colab的Torch版本

  18. 18

    Keras提前停止回调错误,val_loss指标不可用

  19. 19

    如何避免VSCode中的“导入路径不能以.ts扩展名结尾”错误?

  20. 20

    Nuxt.JS:如何在页面中获取路由URL参数

  21. 21

    是否有为什么会AccessibilityManager.sInstance导致内存泄漏的一个原因?

热门标签

归档