将 Hive 结果集转换为多字符分隔的 CSV - SelectHiveQl 处理器 NIFI答案

【问题标题】：Convert Hive result Set to Multi-Character delimited CSV - SelectHiveQl Processor NIFI将 Hive 结果集转换为多字符分隔的 CSV - SelectHiveQl 处理器 NIFI
【发布时间】：2018-06-19 11:07:34
【问题描述】：

我正在尝试使用 selectHiveQL 处理器从 hive 中获取具有 10M+ 记录的大型完整表，并且确实发现源代码中的 converttoCSVStream() 方法比获取结果集花费的时间更长。观察代码：结果集正在逐行迭代，然后添加到输出流中。

当表大小较小时，它会在几秒钟内完成该过程，但当数据很大时，它需要更长的时间。有什么办法可以优化转换吗？我尝试使用 100000/1000/10000/1000 的提取大小。

代码如下：

while (rs.next()) {
        //logger.info("+++++++++++++Inside the While loop+++++++++++++++++");
        if (callback != null) {
            callback.processRow(rs);
        }

        List<String> rowValues = new ArrayList<>(nrOfColumns);

        for (int i = 1; i <= nrOfColumns; i++) {
            final int javaSqlType = meta.getColumnType(i);
            final Object value = rs.getObject(i);

            //logger.info("+++++++++++++Entering the Switch at +++++++++++++++++");
            switch (javaSqlType) {

                case CHAR:
                case LONGNVARCHAR:
                case LONGVARCHAR:
                case NCHAR:
                case NVARCHAR:
                case VARCHAR:
                    String valueString = rs.getString(i);
                    if (valueString != null) {
                        // Removed extra quotes as those are a part of the escapeCsv when required.
                        StringBuilder sb = new StringBuilder();
                        if (outputOptions.isQuote()) {
                            sb.append("\"");
                            if (outputOptions.isEscape()) {
                                sb.append(StringEscapeUtils.escapeCsv(valueString));
                            } else {
                                sb.append(valueString);
                            }
                            sb.append("\"");
                            rowValues.add(sb.toString());
                        } else {
                            if (outputOptions.isEscape()) {
                                rowValues.add(StringEscapeUtils.escapeCsv(valueString));
                            } else {
                                rowValues.add(valueString);
                            }
                        }
                    } else {
                        rowValues.add("");
                    }
                    break;
                case ARRAY:
                case STRUCT:
                case JAVA_OBJECT:
                    String complexValueString = rs.getString(i);
                    if (complexValueString != null) {
                        rowValues.add(StringEscapeUtils.escapeCsv(complexValueString));
                    } else {
                        rowValues.add("");
                    }
                    break;
                default:
                    if (value != null) {
                        rowValues.add(value.toString());
                    } else {
                        rowValues.add("");
                    }
            }
            //logger.info("+++++++++++++Exiting the Switch at +++++++++++++++++" + System.currentTimeMillis());
        }


        // Write row values
        //logger.info("+++++++++++++Writing Row value at+++++++++++++++++" + System.currentTimeMillis());
        outStream.write(StringUtils.join(rowValues, outputOptions.getDelimiter()).getBytes(StandardCharsets.UTF_8));
        outStream.write("\n".getBytes(StandardCharsets.UTF_8));
        nrOfRows++;


        if (maxRows > 0 && nrOfRows == maxRows)
            break;
    }

【问题讨论】：

为什么不在输出流上使用 writer(utf8)？而不是写入stringbuffer，然后使用string.join - 直接写入输出。检查流是否被缓冲...也许您可以将一个查询拆分为多个结果集较小的查询并并行运行。
还要检查是否可以为您的驱动程序设置获取大小issues.apache.org/jira/browse/HIVE-10982
@daggett 此代码 sn-p 来自该工具的 Java 代码。可以设置获取大小，正如我所提到的，我也尝试了不同的获取大小。
tool's java code in place - 你是什么意思？不能修改代码？

标签： java jdbc hive apache-nifi jdbctemplate

【解决方案1】：

NIFI-5307 中介绍了 SelectHiveQL 的 CSV 输出的改进，但尚未实施。也有讨论（但没有我知道的 Jira）允许增量提交（即，在处理完部分结果后立即发送部分结果，而不是在结果集完全处理后发送所有结果），例如在QueryDatabaseTable (NIFI-4836)。

【讨论】：