检测到您已登录华为云国际站账号，为了您更好的体验，建议您访问国际站服务网站 https://www.huaweicloud.com/intl/zh-cn
不再显示此消息
检测到您已登录华为云国际站账号，为了您更好的体验，建议您访问国际站服务网站 https://www.huaweicloud.com/intl/zh-cn
不再显示此消息
基于Kafka的Word Count数据流统计案例 应用场景 Kafka是一个分布式的消息发布-订阅系统。它采用独特的设计提供了类似JMS的特性，主要用于处理活跃的流式数据。
集群在线扩缩容 大数据集群的处理能力通常可以通过增加集群的节点数来横向扩展，当集群规模不符合业务要求时，用户可以通过该功能进行集群节点规模的调整，进行扩容或者缩容；在缩容节点时，MRS会智能地选择负载最少或者迁移数据量最小节点，并且在缩容过程中，缩容节点不再接收新的任务，正在执行的任务继续执行
在线检索MRS集群日志 MRS集群Manager支持在线检索并显示组件的日志内容，用于问题定位等其他日志查看场景，管理员可在线按照节点范围或者组件角色范围快速检视所有日志，通过关键字分析快速定位问题。 本章节操作仅支持MRS 3.x及之后的版本。
spark-beeline 在spark-beeline中执行命令，例如在obs://mrs-word001/table/目录中创建表test。
用户可以在MRS管理控制台在线创建一个作业并提交运行，也可以通过MRS集群客户端来以命令行形式提交作业。 前提条件 用户已经将作业所需的程序包和数据文件上传至OBS或HDFS文件系统中。
this.kvStore.put(word, 1); } else { this.kvStore.put(word, oldValue + 1); }
this.kvStore.put(word, 1); } else { this.kvStore.put(word, oldValue + 1); }
words = lines.select(explode(split(lines.value, " ")).alias("word")) # 生成正在运行的word count wordCounts = words.groupBy("word").count
word.isEmpty()) { word = word.toLowerCase(); collector.emit(new Values(word)); }
word.isEmpty()) { word = word.toLowerCase(); collector.emit(new Values(word)); }
words = lines.select(explode(split(lines.value, " ")).alias("word")) # 生成正在运行的word count wordCounts = words.groupBy("word").count
hdfs dfs -ls obs://OBS并行文件系统名称/路径 例如，执行以下命令访问“mrs-word001”并行文件系统，返回文件列表即表示访问OBS成功，如图1所示： hadoop fs -ls obs://mrs-word001/ 图1 Hadoop验证返回文件列表 使用以下命令上传客户端节点
words = lines.select(explode(split(lines.value, " ")).alias("word")) # 生成正在运行的word count wordCounts = words.groupBy("word").count
this.kvStore.put(word, 1); } else { this.kvStore.put(word, oldValue + 1); }
words = lines.select(explode(split(lines.value, " ")).alias("word")) # 生成正在运行的word count wordCounts = words.groupBy("word").count
代码样例（代码中涉及到的IP端口请修改为实际的IP及端口） public class SimpleJDBCTopology { private static final String WORD_SPOUT = "WORD_SPOUT"; private static
wordSpout); builder.setBolt(JDBC_LOOKUP_BOLT, wordLookupBolt, 1).fieldsGrouping(WORD_SPOUT, new Fields("WORD")); builder.setBolt
wordSpout); builder.setBolt(JDBC_LOOKUP_BOLT, wordLookupBolt, 1).fieldsGrouping(WORD_SPOUT, new Fields("WORD")); builder.setBolt
word.isEmpty()) { word = word.toLowerCase(); collector.emit(new Values(word)); }
") .withColumnFields(new Fields("word")) .withCounterFields(new Fields("count"))
长按/截图保存，微信识别二维码
或者关注公众号“华为云”
