代码之家 › 专栏 › 技术社区 › Kermit

从单个大文件的行创建Kafka事件流

spark-streaming apache-kafka

Kermit · 技术社区 · 7 年前

在dna信息学中,这些文件非常庞大(每个300gb,而biobanks有数十万个文件),它们需要经过6个左右长的下游管道(从几个小时到几个星期)。因为我不在生产测序机的公司工作,所以我不能访问正在生成的数据……也不能编写汇编语言。

我想做的是将300GB文件中的文本行转换为流事件。然后通过6个管道传递这些消息,kafka代理在每个管道之间传递sparkstreaming。

这可能吗?这是错误的用例吗?最好重新运行单个事件,而不是整个失败的批处理。

Desired Workflow:
------pipe1------
_------pipe2------
__------pipe3------
___------pipe4------


Current Workflow:
------pipe1------
_________________------pipe2------
__________________________________------pipe3------
___________________________________________________------pipe4------

1 回复 | 直到 7 年前

OneCricketeer Gabriele Mariotti 7 年前

卡夫卡不是用来发送文件的,只是相对较小的事件。即使您逐行发送文件,您也需要知道如何将文件放回一起进行处理,因此您可以有效地执行与通过原始TCP套接字传输文件相同的操作。

Kafka的最大默认消息suze是1MB,虽然您可以增加它,但我不建议将它推到两位数MB的大小之上。

How can I send large messages with Kafka (over 15MB)?

如果您真的需要通过kafka获得这样的数据,建议的模式是将大文件放在外部存储器(hdfs、s3等)上,然后将uri放在kafka事件中的文件中,并让消费者处理读取该数据源的问题。

如果这些文件有任何结构(例如页面),那么可以使用spark和定制的hadoop inputformat序列化这些文件,并以这种方式并行处理数据。不过,不一定非要通过卡夫卡。你可以试试apache nifi,我听说它能更好地处理更大的文件(可能不是gb)。

推荐文章

xiang qi · 结构化流媒体程序无法连续运行。运行几分钟后,打印[INFO]StateStore:57

2 年前

user3692015 · 流处理如何处理历史聚合?

3 年前

data_pikachu · 班级组织。阿帕奇。hadoop。财政司司长。s3a。啊。尝试从Spark在S3存储桶上写入数据时,未找到IAMInstanceCredentialsProvider

4 年前

sharon gur · Kafka spark同步流处理作业

8 年前

vijay · Spark streaming作业日志大小溢出

8 年前

Pawan · Spark Streaming中setJars()方法的输入应该是什么

8 年前

Frank · Spark Streaming kafka偏移量管理

8 年前

a.moussa · 线程“main”java中出现异常。lang.NoClassDefFoundError:org/apache/spark/streaming/StreamingContext

8 年前

Francesco · 在Spark Streaming中将StreamingContext与接收器分离

8 年前

maxness · Spark将Kafka输入流另存为Json文件

8 年前