flume

0热度

1回答

我想从Flume源文件写入Kafka主题。在水槽中有卡夫卡频道，并在这Cloudera post作者说，可以使用卡夫卡频道要直接从Flume源写入卡夫卡，而无需额外的缓冲。但是，当我试图从我的配置水槽排除汇说在验证这个配置时发生错误：组件tier1.sinks：属性值丢失。我真的需要写信给kafka频道，然后回头再写一遍kafka接收器吗？这对我来说似乎很奇怪......

0热度

1回答

如何将流日志数据追加到Flume中的hdfs文件中？有没有人有MR源代码将数据附加到hdfs

中的文件我需要使用Flume将流数据添加到hdfs中。在不覆盖现有日志文件的情况下，我需要将流数据附加到hdfs中的现有文件。您能否提供相同的MR代码链接？

0热度

1回答

需要一个简单的多线程水槽源

寻找咨询如何建立一个简单的多线程水槽源。我一直在使用spoolDir源在我的本地开发环境中测试我的拦截，但是从下面的链接，它看起来像spoolDir来源永远是单线程： Reading Flume spoolDir in parallel 我们有一个拦截器，可能是生产中的错误，因为它使用非线程安全的SimpleDateFormatter，我试图在我简单的开发环境中重现。

0热度

1回答

FluF冲突hadoop与SLF4J：类路径包含多个SLF4J绑定

当我每次启动Flume时，我都在Flume，Hive和Hadoop之间发送此消息。避免这种情况的最好方法是什么？我正在考虑从flume lib目录中删除一个jar，但不知道是否会影响其他人（配置单元，hadoop）。 Info: Sourcing environment configuration script /usr/local/flume/conf/flume-env.sh Info: I

0热度

1回答

在hdfs kafka，flume的不同目录中着陆文件

我想从一个服务器（Unix服务器A）发送csv文件到hdfs目录。基于哪些csv文件我想将它们放在hdfs的不同目录中。我有11个csv文件将由kafka生产者在unix服务器A上发送到kafka。 csv文件的第一个元素将包含一个键。我想把第一个元素作为kafka消息的关键。另外，我想发送数据的值作为kafka中的消息。当它遇到kafka集群时，会有一个获取数据的通道代理程序和通过密钥对数据

2热度

2回答

对新列使用不同的avro模式

我正在使用flume + kafka将日志数据下载到hdfs。我的接收器数据类型是Avro。在avro模式（.avsc）中，有80个字段作为列。所以我创建了一个外部表像 CREATE external TABLE pgar.tiz_biaws_fraud PARTITIONED BY(partition_date INT) ROW FORMAT SERDE 'org.apache.hadoo

0热度

1回答

IIS日志实时绑定到Hadoop

我正在尝试在Hadoop中执行POC以进行日志聚合。我们有多个托管至少100个站点的IIS服务器。我想要将日志连续传输到HDFS并解析数据并存储在Hive中以进行进一步分析。 1）就是Apache KAFKA正确的选择或Apache水槽 2）流是后能够更好地使用Apache风暴和提取数据到蜂巢如有任何建议，帮助和这种也可以是任何信息问题陈述。感谢

0热度

1回答

Flume需要时间将文件上传到HDFS

我需要得到您的帮助以检查flume为什么需要花费时间将平面文件上传到HDFS。我尝试上载了1个文件（10MB大小），但17小时过去了，它仍然以“.tmp”上传。当我检查了日志详细信息，就好像它卡在通道： 11月10日，X：XX：XX.XXX PM信息org.apache.flume.channel.file.LogFile关闭RandomReader /数据5/flumedata/log-1是 1

0热度

1回答

根据文件大小滚动时，需要花时间将数据复制到hdfs中

我有一个用例，我想用远程文件复制到hdfs使用水槽。我还希望复制的文件应与HDFS块大小（128MB/256MB）一致。远程数据的总大小为33GB。我正在使用avro源和接收器将远程数据复制到hdfs。同样从汇方我做文件大小滚动（128,256）。但从远程机器复制文件并将其存储到hdfs（文件大小128/256 MB）水槽采取平均2分钟。水槽配置：的Avro源（远程计算机） ### Agen

0热度

2回答

带有流水线的结构化流式传输

嗨，有谁能告诉我如何使用Spark全新的API来读取流式流结构化流式传输。实施例： val lines = spark.readStream.format("socket").option("host", "localhost").option("port", 9999).load()