加速 Storm 拓扑答案

【问题标题】：Speeding up Storm Topology加速 Storm 拓扑
【发布时间】：2019-05-24 01:10:58
【问题描述】：

我们有一个将 csv 文件从 HDFS 移动到 Hive 的应用程序。我们正在为该过程使用 Storm 拓扑。

8 台机器一直在使用。它们每个都有 22 个内核和 512 GB RAM。但是，我们的代码运行速度非常慢。传输 600 万条数据需要 10 分钟。

10 MB 的 60 个文件在一秒钟内传输到 HDFS。我们正在尝试优化我们的代码，但很明显我们做错了什么。

对于 Hive 表，我们有 64 个桶。

在我们的拓扑中，我们有 1 个 Spout 和 2 个 Bolts。基本上，我们的 Spout 获取 CSV 文件，向负责解析数据的第一个 Bolt 发出行，然后 Bolt 向负责 HDFS 进程的第二个 Bolt 发出。

HDFS 喷口；

HdfsSpout hdfsSpout = new HdfsSpout()
    .withOutputFields(TextFileReader.defaultFields)
    .setReaderType("text")
    .setHdfsUri(hdfsUri)
    .setSourceDir("/data/in")
    .setArchiveDir("/data/done")
    .setBadFilesDir("/data/bad")
    .setClocksInSync(true) // NTP installed on all hosts
    .setIgnoreSuffix("_COPYING_") 
// do not begin reading file until it is completely copied to HDFS
    .setMaxOutstanding(50_000);

映射器；

DelimitedRecordHiveMapper mapper = new DelimitedRecordHiveMapper()
    .withColumnFields(new Fields(TTDPIRecord.fieldsList))
    .withPartitionFields(new Fields(TTDPIRecord.partitionFieldsList));

蜂巢选项；

HiveOptions hiveOptions = new HiveOptions(metaStoreURI, dbName, tblName, mapper)
    .withAutoCreatePartitions(true)
    .withHeartBeatInterval(3)
    .withCallTimeout(10_000) // default = 10.000
    .withTxnsPerBatch(2)
    .withBatchSize(50_000) 
// doing below because its affecting storm metrics most likely
    .withTickTupleInterval(1);

配置；

Config conf = new Config();
conf.setNumWorkers(6);
conf.setNumAckers(6);
conf.registerMetricsConsumer(LoggingMetricsConsumer.class);

拓扑生成器；

TopologyBuilder builder = new TopologyBuilder();
builder.setSpout("hdfsSpout", hdfsSpout, 8);
builder.setBolt("recordParserBolt", recordParserBolt, 8).localOrShuffleGrouping("hdfsSpout");
builder.setBolt("hiveBolt", hiveBolt, 8).localOrShuffleGrouping("recordParserBolt");

我们不确定以下参数；

在 HDFS Spout 中； .setMaxOutstanding(50_000);

在 Hive Spout 选项中； .withTxnsPerBatch(2) .withBatchSize(50_000) .withTickTupleInterval(1);

在配置中； .setNumWorkers(6); .setNumAckers(6);

Spout 和 Bolt 中的平行度；我们给了每个 8 个。

这些参数的值应该是多少？提前致谢。

编辑；这是我们对 100 个 csv 文件的 10 MB 的测试结果；

hdfsSpout 执行者：8 完整延迟：1834.209 毫秒

recordParserBolt 执行器：8 完整延迟：0.019 毫秒

hiveBolt 执行器：8 完整延迟：1092.624 毫秒

【问题讨论】：

标签： hive apache-storm topology bolt spout

【解决方案1】：

您正在使用conf.setNumWorkers(6);，这意味着您只使用了 8 台机器中的 6 台，您可以将其设置为 8 以利用您拥有的所有硬件。

您可以更改的另一个参数是螺栓的并行性提示，这意味着组件的执行器（线程）的初始数量。您只给了 8 个并行度，您可以将其增加到 100/200，看看性能如何变化。

您可以通过this 了解 Storm 中的并行工作原理。

你能告诉你max-spout-pending的配置是什么吗？

【讨论】：