大数据开发 – 第24页

Hive, Spark 5月 27,2022

Idea用Spark SQL远程访问CDH6.3.2的hive

win10的开发环境，想在idea用Spark SQL来访问CDH6.3.2服务器的hive。刚开始用pom配置原生的spark等jar包。发现访问有问题，各种折腾还是没解决。

看到网上说要用cdh的包。就把相关的包替换成cdh的包。没想到用cdh的包各种缺少依赖包，折腾几个小时还没解决。

后来想到一个简单暴力的办法，把cdh集群的的jar包都下载下来，然后idea添加这个文件夹为依赖。

果然解决缺少依赖包的问题，不用1个个去排查添加了。不过新问题随之而来。又提示ANTLR Tool version 4.5 used for code generation does not match the current runtime version 4.7ANTLR

按网上教程处理未果，想到会不会是CDH的jar包版本重复导致的，一看果然是这样，先把antlr4-runtime-4.5.1-1.jar和antlr4-runtime-4.5.jar排除掉。重新运行果然是可以了。

还需要做下面的操作：

在本地安装spark，并把远程服务器hive.xml、core-site.xml、mapred-site.xml和yarn-site.xml拷贝到spark的conf

如果远程服务器是有hostname的，需要修改hosts

作者 east

Hive 5月 24,2022

用Sqoop导入mysql到hive遇到的坑

坑1：所有字段的值都为null

发现原因是在于，建hive表是设定的分割符不恰当，跟从mysql导入过来的数据的分隔符不一样，所以导致hive切分不了数据，于是查询为空，但是这个过程，是不属于导入失败的，所以导入命令没有报错。因为sqoop import实际上是把数据存放到hdfs对应路径上了，而不是“直接导入表里”，查询时，hive会从hdfs的路径上提取数据，再根据hive表的结构和定义，来向我们展示出类似表格的形式。因此，导入过程是不会报错的，但是因为hive定义的分隔符和存在hdfs上数据的分隔符不一致，所以查询是全为NULL的。查看自己hive的建表语句

CREATE TABLE IF NOT EXISTS ods.test1 (
id BIGINT, 
type_id INT, 
parent_id INT,
 name STRING, 
 note_state INT) 
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\t' STORED AS TEXTFILE;

可以看到分隔符为：FIELDS TERMINATED BY ‘\t’，而从postgresql或者mysql来的数据的分隔符则应该为：FIELDS TERMINATED BY ‘\u0001’，那我们只要改回来就可以正常导入了。

把表删了,重新建表,指定分隔符为FIELDS TERMINATED BY ‘\u0001’.

CREATE TABLE IF NOT EXISTS ods.test1 (
id BIGINT, 
type_id INT, 
parent_id INT,
 name STRING, 
 note_state INT) 
ROW FORMAT DELIMITED FIELDS TERMINATED BY '\u0001' STORED AS TEXTFILE;

坑2： tinyint类型的都为null

sqoop运⾏完成之后，发现为tinyint类型的⼀类始终为null，经检查发现上游mysql有值，再查看hdfs⽂件，发现这列被抓换为了boolean类型。解决方案：在jdbc的连接后⾯加上：tinyInt1isBit=false，例如connect jdbc:mysql://192.168.1.80:3306/my_log?tinyInt1isBit=false

作者 east

Android, spring 4月 29,2022

生产环境app和后台都看不到日志的异常调试方法

最近调试一个项目，调试到让人亏崩溃。

由于是特殊行业，手机不开usb调试模式，这样没法看到android app的日志。由于疫情等原因，又无法到现场看后台日志。在调试app登录时，app一直报“ 网络异常，请稍后重试” 。

由于网络是经过第三方转发的，刚开始以为是第三方转发那方面没配置好。经过费劲周折，先后发现app的key、secret、url配错了，又测了后台直接调用的接口是通的，反复找第三方转发公司确认，才确认第三方转发是正确的，整个网络是通的。

全面的测试用例查找问题：

app端对网络异常处理不够完善，当后台500出错时，app没准确提示，还是提示统一的“网络异常，请稍后重试”。怀疑app访问超时导致这样问题，修改超时时间还是没返回。后来别的同事加入讨论，他测试的账号是不存在的，终于有了后台提示。我之前一直是用正确存在的账号测试，反而一直没有收到后台返回结果。看来测试还是要考虑正常和非正常等多元化测试，考虑全面的测试用例，这样可以覆盖代码多个分支。

替代排除法：

通过上面的步骤，已经确认了app能正确访问到生产环境的后台，但不确定是app和后台都有问题，还只是后台有问题。想到生产环境还有一套接口一模一样生产环境的后台，是能正常使用的。把app的url修改成那个后台的地址，发现app登录等一切正常。

通过日志分析：

没办法调试app和后台，但日志还是可以想办法拿到的。android的日志可以上传到后台，也可以保存到本地，通过文件浏览器来找到日志文件。例如在Android把调试日志保存到本地，然后在手机找到安装包包名对应的目录，就可以查找到日志。

通过上面的方法，看到app日志报的是500错误，那么是后台接口报错，后来通过查看后台日志，果然看到空指针错误了。处理好问题了，果然app能正常登录后台和正常使用了。

作者 east

Flink, Spark 4月 19,2022

大数据实时流式处理:Apache Flink vs Apache

对更快数据处理的需求一直在增加，实时流数据处理是目前的解决方案。虽然 Apache Spark 仍在许多组织中用于大数据处理，但 Apache Flink 已经迅速成为替代方案。事实上，许多人认为它有可能取代 Apache Spark，因为它能够实时处理流数据。当然，Flink 能否取代 Spark 尚无定论，因为 Flink 还没有经过广泛的测试。但实时处理和低数据延迟是其两个决定性特征。同时，这需要考虑到 Apache Spark 可能不会失宠，因为它的批处理能力仍然很重要。

流式数据处理案例

对于基于批处理的所有优点，实时流数据处理似乎是一个强有力的案例。流式数据处理使快速设置和加载数据仓库成为可能。具有低数据延迟的流处理器可以快速提供对数据的更多见解。所以，你有更多的时间来了解发生了什么。除了更快的处理之外，还有另一个显着的好处：您有更多的时间来设计对事件的适当响应。例如，在异常检测的情况下，更低的延迟和更快的检测使您能够确定最佳响应，这是防止安全网站受到欺诈攻击或工业设备损坏等情况的关键。因此，您可以防止重大损失。

什么是 Apache Flink？

Apache Flink 是一种大数据处理工具，以在大规模分布式系统上以低数据延迟和高容错性快速处理大数据而著称。它的定义特征是它能够实时处理流数据。

Apache Flink 最初是一个学术开源项目，当时它被称为 Stratosphere。后来，它成为了 Apache 软件基金会孵化器的一部分。为避免与其他项目名称冲突，将名称更改为 Flink。 Flink 这个名字很合适，因为它意味着敏捷。即使选择的标志，松鼠也是合适的，因为松鼠代表了敏捷、敏捷和速度的美德。

自从加入 Apache 软件基金会后，它作为大数据处理工具迅速崛起，并在 8 个月内开始受到更广泛受众的关注。人们对 Flink 的兴趣日益浓厚，这反映在 2015 年的多次会议的参会人数上。2015 年 5 月在伦敦举行的 Strata 会议和 2015 年 6 月在圣何塞举行的 Hadoop 峰会上，有很多人参加了关于 Flink 的会议。 2015 年 8 月，超过 60 人参加了在圣何塞 MapR 总部举办的湾区 Apache Flink 聚会。

下图给出了 Flink 的 Lambda 架构。

Spark 和 Flink 的比较

虽然 Spark 和 Flink 之间有一些相似之处，例如它们的 API 和组件，但在数据处理方面，相似之处并不重要。下面给出了 Flink 和 Spark 之间的比较。

数据处理

Spark 以批处理模式处理数据，而 Flink 实时处理流数据。 Spark 处理数据块，称为 RDD，而 Flink 可以实时处理一行一行的数据。因此，虽然 Spark 始终存在最小数据延迟，但 Flink 并非如此。

迭代

Spark 支持批量数据迭代，但 Flink 可以使用其流式架构原生迭代其数据。下图显示了迭代处理是如何发生的。

内存管理

Flink 可以自动适应不同的数据集，但 Spark 需要手动优化和调整其作业以适应单个数据集。 Spark 也进行手动分区和缓存。因此，预计处理会有所延迟。

数据流

Flink 能够在需要时为其数据处理提供中间结果。 Spark 遵循过程式编程系统，而 Flink 遵循分布式数据流方法。因此，当需要中间结果时，广播变量用于将预先计算的结果分发到所有工作节点。

数据可视化

Flink 提供了一个 Web 界面来提交和执行所有作业。 Spark 和 Flink 都与 Apache Zeppelin 集成，并提供数据摄取、数据分析、发现、协作和可视化。 Apache Zeppelin 还提供了多语言后端，允许您提交和执行 Flink 程序。

处理时间

以下段落提供了 Flink 和 Spark 在不同作业中所用时间的比较。

为了公平比较，Flink 和 Spark 都以机器规格和节点配置的形式获得了相同的资源。

Flink 处理速度更快，因为它的流水线执行。处理数据，Spark 用了 2171 秒，而 Flink 用了 1490 秒。

当执行不同数据大小的 TeraSort 时，结果如下：

对于 10 GB 的数据，Flink 需要 157 秒，而 Spark 需要 387 秒。
对于 160 GB 的数据，Flink 需要 3127 秒，而 Spark 需要 4927 秒。
基于批处理或流式数据——哪个过程更好？

这两种工艺各有优势，适用于不同的情况。尽管许多人声称基于批处理的工具正在失宠，但它不会很快发生。要了解它们的相对优势，请参见以下比较：

在个别情况下，Flink 和 Spark批处理都是有用的。以每天计算滚动月销售额的用例为例。在此活动中，需要计算每日销售总额，然后进行累计。在这样的用例中，可能不需要对数据进行流式处理。数据的批处理可以根据日期处理各个批次的销售数据，然后将它们添加。在这种情况下，即使存在一些数据延迟，也可以在稍后将该潜在数据添加到以后的批次中时弥补这些延迟。

有类似的用例需要流处理。以计算每个访问者在网站上花费的每月滚动时间的用例为例。在网站的情况下，访问次数可以每小时、每分钟甚至每天更新一次。但是这种情况下的问题是定义会话。定义会话的开始和结束可能很困难。此外，难以计算或识别不活动的时间段。因此，在这种情况下，定义会话甚至不活动时间段都没有合理的界限。在这种情况下，需要实时处理流数据。

概括

虽然 Spark 在批处理数据处理方面有很多优势，而且它仍然有很多使用场景，但 Flink 似乎正在迅速获得商业方面应用的青睐。 Flink 也可以进行批处理这一事实似乎对其有利。当然，这需要考虑到 Flink 的批处理能力可能与 Spark 不在一个级别。

作者 east

Java 4月 16,2022

网格编号工具类


import java.math.BigDecimal;

/**
 * 网格编号工具类<br>
 * 一、业务数据坐标网格编号计算 <br>
 * 1）基于覆盖全国的要求，地理坐标以左下角为原点，采用经度72°及纬度 0°为网格坐标原点进行网格划分<br>
 * 2）根据不同的比例尺要求，将网格切分成不同的级别，并按照4x4进行逐级进行切割，精度保留到小数点后10位<br>
 * 3）从原点开始计算（原点处坐标为72°,0°），根据网格范围进行行列号换算，其中一次网格行号和列号采用1位或2位数据表示（ 0,1,2,3,4…），其他均采用1位数表达（0,1,2,3）<br>
 * 二、计算最小网格的中心点坐标<br>
 * 1)通过入参确定需要计算的最小网格中心点坐标 <br>
 * 2）先计算出一次网格的跨度，如果有更小的网格则累加起来，并取最后一个网格的半个跨度
 * 
 */
public final class GridCodeUtil {

    /** 网格化原点经度72° */
    public static final double ORIGIN_X = 72d;

    /** 网格化原点纬度 0° */
    public static final double ORIGIN_Y = 0d;

    /** 网格化终点经度 136° */
    public static final double END_X = 136d;

    /** 网格化终点纬度 54° */
    private static final double END_Y = 54d;

    /** 网格化无效值 */
    public static final int INVALID_VALUE = -1;

    /** 一次网格大小，即经度 1度，纬度1度 */
    public static final double FIRST_GRID_SIZE = 1d;

    /** 二次网格大小，即经度0.25度，纬度0.25度 */
    public static final double SECOND_GRID_SIZE = 0.25d;

    /** 三次网格大小，即经度0.0625度，纬度0.0625度 */
    public static final double THIRD_GRID_SIZE = 0.0625d;

    /** 四次网格大小，即经度0.015625度，纬度0.015625度 */
    public static final double FOURTH_GRID_SIZE = 0.015625d;

    /** 五次网格大小，即经度0.00390625度，纬度0.00390625度 */
    public static final double FIFTH_GRID_SIZE = 0.00390625d;

    /** 六次网格大小，即经度0.0009765625度，纬度0.0009765625度 */
    public static final double SIXTH_GRID_SIZE = 0.0009765625d;

    /** 一次网格折半大小 */
    public static final double FIRST_GRID_HALF_SIZE = FIRST_GRID_SIZE / 2;

    /** 二次网格折半大小 */
    public static final double SECOND_GRID_HALF_SIZE = SECOND_GRID_SIZE / 2;

    /** 三次网格折半大小 */
    public static final double THIRD_GRID_HALF_SIZE = THIRD_GRID_SIZE / 2;

    /** 四次网格折半大小 */
    public static final double FOURTH_GRID_HALF_SIZE = FOURTH_GRID_SIZE / 2;

    /** 五次网格折半大小 */
    public static final double FIFTH_GRID_HALF_SIZE = FIFTH_GRID_SIZE / 2;

    /** 六次网格折半大小 */
    public static final double SIXTH_GRID_HALF_SIZE = SIXTH_GRID_SIZE / 2;

    /** 私有构造函数 */
    private GridCodeUtil() {
    }

    /**
     * 坐标网格编号计算
     * 
     * @param xcoord 点的x坐标值，即经度
     * @param ycoord 点的y坐标值，即纬度
     * @return
     */
    public static GridCode computeGridCode(Double xcoord, Double ycoord) {
        GridCode result = new GridCode();
        boolean isNullOrOutOfRange4x = xcoord == null || xcoord < ORIGIN_X || xcoord > END_X;
        boolean isNullOrOutOfRange4y = ycoord == null || ycoord < ORIGIN_Y || ycoord > END_Y;

        // 校验是否合法的坐标
        if (isNullOrOutOfRange4x || isNullOrOutOfRange4y) {
            result.setInvalidValue(INVALID_VALUE);
        } else {
            result.setFirstGridRowCode(computeRowCodeOfFirstGrid(ycoord));
            result.setFirstGridColCode(computeColumnCodeOfFirstGrid(xcoord));
            result.setSecondGridRowCode(computeRowCodeOfSecondGrid(ycoord));
            result.setSecondGridColCode(computeColumnCodeOfSecondGrid(xcoord));
            result.setThirdGridRowCode(computeRowCodeOfThirdGrid(ycoord));
            result.setThirdGridColCode(computeColumnCodeOfThirdGrid(xcoord));
            result.setFourthGridRowCode(computeRowCodeOfFourthGrid(ycoord));
            result.setFourthGridColCode(computeColumnCodeOfFourthGrid(xcoord));
            result.setFifthGridRowCode(computeRowCodeOfFifthGrid(ycoord));
            result.setFifthGridColCode(computeColumnCodeOfFifthGrid(xcoord));
            result.setSixthGridRowCode(computeRowCodeOfSixthGrid(ycoord));
            result.setSixthGridColCode(computeColumnCodeOfSixthGrid(xcoord));
        }
        return result;
    }

    /**
     * 计算最小网格的中心点坐标
     * 
     * @param gridCode 网格编号对象
     * @return 中心点坐标，数组的第一个值是x坐标，第二个值是y坐标
     */
    public static Double[] computeCentrePointOfMinGrid(GridCode gridCode) {
        Double xcoord = null;
        Double ycoord = null;
        int minGrid = 0;

        // 先计算出一次网格的跨度，如果有更小的网格则累加起来。因网格计算时，编号是从0开始的，因此不需要减去一
        if (gridCode.getFirstGridColCode() != null && gridCode.getFirstGridRowCode() != null) {
            xcoord = gridCode.getFirstGridColCode() * FIRST_GRID_SIZE;
            ycoord = gridCode.getFirstGridRowCode() * FIRST_GRID_SIZE;
            minGrid = 1;
        } else {
            // 如果连最基本的一次网格编号都为空，直接抛出参数异常
            throw new IllegalArgumentException("计算最小网格的中心点坐标时，传入的一次网格编号不能为空！");
        }

        if (gridCode.getSecondGridColCode() != null && gridCode.getSecondGridRowCode() != null) {
            xcoord += gridCode.getSecondGridColCode() * SECOND_GRID_SIZE;
            ycoord += gridCode.getSecondGridRowCode() * SECOND_GRID_SIZE;
            minGrid = 2;
        }

        if (gridCode.getThirdGridColCode() != null && gridCode.getThirdGridRowCode() != null) {
            // 如果二次网格编号为空，直接抛出参数异常
            if (minGrid != 2) {
                throw new IllegalArgumentException("计算最小网格的中心点坐标时，传入的二次网格编号不能为空！");
            }
            xcoord += gridCode.getThirdGridColCode() * THIRD_GRID_SIZE;
            ycoord += gridCode.getThirdGridRowCode() * THIRD_GRID_SIZE;
            minGrid = 3;
        }

        if (gridCode.getFourthGridColCode() != null && gridCode.getFourthGridRowCode() != null) {
            // 如果三次网格编号为空，直接抛出参数异常
            if (minGrid != 3) {
                throw new IllegalArgumentException("计算最小网格的中心点坐标时，传入的三次网格编号不能为空！");
            }
            xcoord += gridCode.getFourthGridColCode() * FOURTH_GRID_SIZE;
            ycoord += gridCode.getFourthGridRowCode() * FOURTH_GRID_SIZE;
            minGrid = 4;
        }

        if (gridCode.getFifthGridColCode() != null && gridCode.getFifthGridRowCode() != null) {
            // 如果四次网格编号为空，直接抛出参数异常
            if (minGrid != 4) {
                throw new IllegalArgumentException("计算最小网格的中心点坐标时，传入的四次网格编号不能为空！");
            }
            xcoord += gridCode.getFifthGridColCode() * FIFTH_GRID_SIZE;
            ycoord += gridCode.getFifthGridRowCode() * FIFTH_GRID_SIZE;
            minGrid = 5;
        }

        if (gridCode.getSixthGridColCode() != null && gridCode.getSixthGridRowCode() != null) {
            // 如果五次网格编号为空，直接抛出参数异常
            if (minGrid != 5) {
                throw new IllegalArgumentException("计算最小网格的中心点坐标时，传入的五次网格编号不能为空！");
            }
            xcoord += gridCode.getSixthGridColCode() * SIXTH_GRID_SIZE;
            ycoord += gridCode.getSixthGridRowCode() * SIXTH_GRID_SIZE;
            minGrid = 6;
        }
        return buildResult(xcoord, ycoord, minGrid);
    }

    /**
     * 计算指定的网格的中心点坐标
     *
     * @param gridCode 网格编号对象
     * @param minGrid 要计算的网格层级
     * @return 中心点坐标，数组的第一个值是x坐标，第二个值是y坐标
     */
    public static Double[] computeCentrePointOfMinGrid2(GridCode gridCode,int minGrid) {
        Double xcoord = null;
        Double ycoord = null;

        if (minGrid == 1) {
            xcoord = gridCode.getFirstGridColCode() * FIRST_GRID_SIZE;
            ycoord = gridCode.getFirstGridRowCode() * FIRST_GRID_SIZE;
        }

        if (minGrid == 2) {
            xcoord += gridCode.getSecondGridColCode() * SECOND_GRID_SIZE;
            ycoord += gridCode.getSecondGridRowCode() * SECOND_GRID_SIZE;
        }

        if (minGrid == 3) {
            xcoord += gridCode.getThirdGridColCode() * THIRD_GRID_SIZE;
            ycoord += gridCode.getThirdGridRowCode() * THIRD_GRID_SIZE;
        }

        if (minGrid == 4) {
            xcoord += gridCode.getFourthGridColCode() * FOURTH_GRID_SIZE;
            ycoord += gridCode.getFourthGridRowCode() * FOURTH_GRID_SIZE;
        }

        if (minGrid == 5) {
            xcoord += gridCode.getFifthGridColCode() * FIFTH_GRID_SIZE;
            ycoord += gridCode.getFifthGridRowCode() * FIFTH_GRID_SIZE;
            minGrid = 5;
        }

        if (minGrid == 6) {
            xcoord += gridCode.getSixthGridColCode() * SIXTH_GRID_SIZE;
            ycoord += gridCode.getSixthGridRowCode() * SIXTH_GRID_SIZE;
        }
        try {
            return buildResult(xcoord, ycoord, minGrid);
        } catch (Exception e) {
            return null;
        }
    }

    /**
     * 构建最后的结果值<br>
     * 添加网格化原点经纬度和最小网格的中点跨度，如果小数点大于八位，则四舍五入保留八位小数点
     * 
     * @param xcoord 网格跨度x坐标值
     * @param ycoord 网格跨度y坐标值
     * @param minGrid 需要计算的最小网格
     * @return
     */
    private static Double[] buildResult(Double xcoord, Double ycoord, int minGrid) {
        Double[] result = new Double[2];

        result[0] = ORIGIN_X + xcoord;
        result[1] = ORIGIN_Y + ycoord;

        switch (minGrid) {
        case 1:
            result[0] += FIRST_GRID_HALF_SIZE;
            result[1] += FIRST_GRID_HALF_SIZE;
            break;
        case 2:
            result[0] += SECOND_GRID_HALF_SIZE;
            result[1] += SECOND_GRID_HALF_SIZE;
            break;
        case 3:
            result[0] += THIRD_GRID_HALF_SIZE;
            result[1] += THIRD_GRID_HALF_SIZE;
            break;
        case 4:
            result[0] += FOURTH_GRID_HALF_SIZE;
            result[1] += FOURTH_GRID_HALF_SIZE;
            break;
        case 5:
            result[0] += FIFTH_GRID_HALF_SIZE;
            result[1] += FIFTH_GRID_HALF_SIZE;
            break;
        case 6:
            result[0] += SIXTH_GRID_HALF_SIZE;
            result[1] += SIXTH_GRID_HALF_SIZE;
            break;
        default:
            throw new IllegalArgumentException("计算最小网格的中心点坐标异常，当前只支持1-6次网格！");
        }
        BigDecimal bd0 = new BigDecimal(result[0]);
        BigDecimal bd1 = new BigDecimal(result[1]);
        result[0] = bd0.setScale(8, BigDecimal.ROUND_HALF_UP).doubleValue();
        result[1] = bd1.setScale(8, BigDecimal.ROUND_HALF_UP).doubleValue();
        return result;
    }

    /**
     * 计算一次网格的行编号
     * 
     * @param ycoord 点的y坐标值，即纬度
     * @return
     */
    private static int computeRowCodeOfFirstGrid(double ycoord) {
        return (int) (difOriginY(ycoord) / FIRST_GRID_SIZE);
    }

    /**
     * 计算一次网格的列编号
     * 
     * @param xcoord 点的x坐标值，即经度
     * @return
     */
    private static int computeColumnCodeOfFirstGrid(double xcoord) {
        return (int) (difOriginX(xcoord) / FIRST_GRID_SIZE);
    }

    /**
     * 获取当前一次网格最小y坐标值
     * 
     * @param ycoord 点的y坐标值，即纬度
     * @return
     */
    private static double getMiniYcoodOfFirstGrid(double ycoord) {
        return computeRowCodeOfFirstGrid(ycoord) * FIRST_GRID_SIZE;
    }

    /**
     * 获取当前一次网格最小x坐标值
     * 
     * @param xcoord 点的x坐标值，即经度
     * @return
     */
    private static double getMiniXcoodOfFirstGrid(double xcoord) {
        return computeColumnCodeOfFirstGrid(xcoord) * FIRST_GRID_SIZE;
    }

    /**
     * 计算二次网格的行编号
     * 
     * @param ycoord 点的y坐标值，即纬度
     * @return
     */
    private static int computeRowCodeOfSecondGrid(double ycoord) {
        return (int) ((difOriginY(ycoord) - getMiniYcoodOfFirstGrid(ycoord)) / SECOND_GRID_SIZE);
    }

    /**
     * 计算二次网格的列编号
     * 
     * @param xcoord 点的x坐标值，即经度
     * @return
     */
    private static int computeColumnCodeOfSecondGrid(double xcoord) {
        return (int) ((difOriginX(xcoord) - getMiniXcoodOfFirstGrid(xcoord)) / SECOND_GRID_SIZE);
    }

    /**
     * 获取当前二次网格最小y坐标值
     * 
     * @param ycoord 点的y坐标值，即纬度
     * @return
     */
    private static double getMiniYcoodOfSecondGrid(double ycoord) {
        return computeRowCodeOfSecondGrid(ycoord) * SECOND_GRID_SIZE;
    }

    /**
     * 获取当前二次网格最小x坐标值
     * 
     * @param xcoord 点的x坐标值，即经度
     * @return
     */
    private static double getMiniXcoodOfSecondGrid(double xcoord) {
        return computeColumnCodeOfSecondGrid(xcoord) * SECOND_GRID_SIZE;
    }

    /**
     * 计算三次网格的行编号
     * 
     * @param ycoord 点的y坐标值，即纬度
     * @return
     */
    private static int computeRowCodeOfThirdGrid(double ycoord) {
        return (int) ((difOriginY(ycoord) - getMiniYcoodOfFirstGrid(ycoord) - getMiniYcoodOfSecondGrid(ycoord))
                / THIRD_GRID_SIZE);
    }

    /**
     * 计算三次网格的列编号
     * 
     * @param xcoord 点的x坐标值，即经度
     * @return
     */
    private static int computeColumnCodeOfThirdGrid(double xcoord) {
        return (int) ((difOriginX(xcoord) - getMiniXcoodOfFirstGrid(xcoord) - getMiniXcoodOfSecondGrid(xcoord))
                / THIRD_GRID_SIZE);
    }

    /**
     * 获取当前三次网格最小y坐标值
     * 
     * @param ycoord 点的y坐标值，即纬度
     * @return
     */
    private static double getMiniYcoodOfThirdGrid(double ycoord) {
        return computeRowCodeOfThirdGrid(ycoord) * THIRD_GRID_SIZE;
    }

    /**
     * 获取当前三次网格最小x坐标值
     * 
     * @param xcoord 点的x坐标值，即经度
     * @return
     */
    private static double getMiniXcoodOfThirdGrid(double xcoord) {
        return computeColumnCodeOfThirdGrid(xcoord) * THIRD_GRID_SIZE;
    }

    /**
     * 计算四次网格的行编号
     * 
     * @param ycoord 点的y坐标值，即纬度
     * @return
     */
    private static int computeRowCodeOfFourthGrid(double ycoord) {
        return (int) ((difOriginY(ycoord) - getMiniYcoodOfFirstGrid(ycoord) - getMiniYcoodOfSecondGrid(ycoord)
                - getMiniYcoodOfThirdGrid(ycoord)) / FOURTH_GRID_SIZE);
    }

    /**
     * 计算四次网格的列编号
     * 
     * @param xcoord 点的x坐标值，即经度
     * @return
     */
    private static int computeColumnCodeOfFourthGrid(double xcoord) {
        return (int) ((difOriginX(xcoord) - getMiniXcoodOfFirstGrid(xcoord) - getMiniXcoodOfSecondGrid(xcoord)
                - getMiniXcoodOfThirdGrid(xcoord)) / FOURTH_GRID_SIZE);
    }

    /**
     * 获取当前四次网格最小y坐标值
     * 
     * @param ycoord 点的y坐标值，即纬度
     * @return
     */
    private static double getMiniYcoodOfFourthGrid(double ycoord) {
        return computeRowCodeOfFourthGrid(ycoord) * FOURTH_GRID_SIZE;
    }

    /**
     * 获取当前四次网格最小x坐标值
     * 
     * @param xcoord 点的x坐标值，即经度
     * @return
     */
    private static double getMiniXcoodOfFourthGrid(double xcoord) {
        return computeColumnCodeOfFourthGrid(xcoord) * FOURTH_GRID_SIZE;
    }

    /**
     * 计算五次网格的行编号
     * 
     * @param ycoord 点的y坐标值，即纬度
     * @return
     */
    private static int computeRowCodeOfFifthGrid(double ycoord) {
        return (int) ((difOriginY(ycoord) - getMiniYcoodOfFirstGrid(ycoord) - getMiniYcoodOfSecondGrid(ycoord)
                - getMiniYcoodOfThirdGrid(ycoord) - getMiniYcoodOfFourthGrid(ycoord)) / FIFTH_GRID_SIZE);
    }

    /**
     * 计算五次网格的列编号
     * 
     * @param xcoord 点的x坐标值，即经度
     * @return
     */
    private static int computeColumnCodeOfFifthGrid(double xcoord) {
        return (int) ((difOriginX(xcoord) - getMiniXcoodOfFirstGrid(xcoord) - getMiniXcoodOfSecondGrid(xcoord)
                - getMiniXcoodOfThirdGrid(xcoord) - getMiniXcoodOfFourthGrid(xcoord)) / FIFTH_GRID_SIZE);
    }

    /**
     * 获取当前五次网格最小y坐标值
     * 
     * @param ycoord 点的y坐标值，即纬度
     * @return
     */
    private static double getMiniYcoodOfFifthGrid(double ycoord) {
        return computeRowCodeOfFifthGrid(ycoord) * FIFTH_GRID_SIZE;
    }

    /**
     * 获取当前五次网格最小x坐标值
     * 
     * @param xcoord 点的x坐标值，即经度
     * @return
     */
    private static double getMiniXcoodOfFifthGrid(double xcoord) {
        return computeColumnCodeOfFifthGrid(xcoord) * FIFTH_GRID_SIZE;
    }

    /**
     * 计算六次网格的行编号
     * 
     * @param ycoord 点的y坐标值，即纬度
     * @return
     */
    private static int computeRowCodeOfSixthGrid(double ycoord) {
        return (int) ((difOriginY(ycoord) - getMiniYcoodOfFirstGrid(ycoord) - getMiniYcoodOfSecondGrid(ycoord)
                - getMiniYcoodOfThirdGrid(ycoord) - getMiniYcoodOfFourthGrid(ycoord) - getMiniYcoodOfFifthGrid(ycoord))
                / SIXTH_GRID_SIZE);
    }

    /**
     * 计算六次网格的列编号
     * 
     * @param xcoord 点的x坐标值，即经度
     * @return
     */
    private static int computeColumnCodeOfSixthGrid(double xcoord) {
        return (int) ((difOriginX(xcoord) - getMiniXcoodOfFirstGrid(xcoord) - getMiniXcoodOfSecondGrid(xcoord)
                - getMiniXcoodOfThirdGrid(xcoord) - getMiniXcoodOfFourthGrid(xcoord) - getMiniXcoodOfFifthGrid(xcoord))
                / SIXTH_GRID_SIZE);
    }

    /**
     * 目标点和原点的纬度差值
     * 
     * @param ycoord 点的y坐标值，即纬度
     * @return
     */
    private static double difOriginY(double ycoord) {
        return ycoord - ORIGIN_Y;
    }

    /**
     * 目标点和原点的经度差值
     * 
     * @param xcoord 点的x坐标值，即经度
     * @return
     */
    private static double difOriginX(double xcoord) {
        return xcoord - ORIGIN_X;
    }

}

public class GridCode {

    /** 一次网格行号 */
    private Integer firstGridRowCode;

    /** 一次网格列号 */
    private Integer firstGridColCode;

    /** 二次网格行号 */
    private Integer secondGridRowCode;

    /** 二次网格列号 */
    private Integer secondGridColCode;

    /** 三次网格行号 */
    private Integer thirdGridRowCode;

    /** 三次网格列号 */
    private Integer thirdGridColCode;

    /** 四次网格行号 */
    private Integer fourthGridRowCode;

    /** 四次网格列号 */
    private Integer fourthGridColCode;

    /** 五次网格行号 */
    private Integer fifthGridRowCode;

    /** 五次网格列号 */
    private Integer fifthGridColCode;

    /** 六次网格行号 */
    private Integer sixthGridRowCode;

    /** 六次网格列号 */
    private Integer sixthGridColCode;

    /**
     * 设置无效值
     */
    public void setInvalidValue(int invalidValue) {
        this.firstGridRowCode = invalidValue;
        this.firstGridColCode = invalidValue;
        this.secondGridRowCode = invalidValue;
        this.secondGridColCode = invalidValue;
        this.thirdGridRowCode = invalidValue;
        this.thirdGridColCode = invalidValue;
        this.fourthGridRowCode = invalidValue;
        this.fourthGridColCode = invalidValue;
        this.fifthGridRowCode = invalidValue;
        this.fifthGridColCode = invalidValue;
        this.sixthGridRowCode = invalidValue;
        this.sixthGridColCode = invalidValue;
    }

    public Integer getFirstGridRowCode() {
        return firstGridRowCode;
    }

    public void setFirstGridRowCode(Integer firstGridRowCode) {
        this.firstGridRowCode = firstGridRowCode;
    }

    public Integer getFirstGridColCode() {
        return firstGridColCode;
    }

    public void setFirstGridColCode(Integer firstGridColCode) {
        this.firstGridColCode = firstGridColCode;
    }

    public Integer getSecondGridRowCode() {
        return secondGridRowCode;
    }

    public void setSecondGridRowCode(Integer secondGridRowCode) {
        this.secondGridRowCode = secondGridRowCode;
    }

    public Integer getSecondGridColCode() {
        return secondGridColCode;
    }

    public void setSecondGridColCode(Integer secondGridColCode) {
        this.secondGridColCode = secondGridColCode;
    }

    public Integer getThirdGridRowCode() {
        return thirdGridRowCode;
    }

    public void setThirdGridRowCode(Integer thirdGridRowCode) {
        this.thirdGridRowCode = thirdGridRowCode;
    }

    public Integer getThirdGridColCode() {
        return thirdGridColCode;
    }

    public void setThirdGridColCode(Integer thirdGridColCode) {
        this.thirdGridColCode = thirdGridColCode;
    }

    public Integer getFourthGridRowCode() {
        return fourthGridRowCode;
    }

    public void setFourthGridRowCode(Integer fourthGridRowCode) {
        this.fourthGridRowCode = fourthGridRowCode;
    }

    public Integer getFourthGridColCode() {
        return fourthGridColCode;
    }

    public void setFourthGridColCode(Integer fourthGridColCode) {
        this.fourthGridColCode = fourthGridColCode;
    }

    public Integer getFifthGridRowCode() {
        return fifthGridRowCode;
    }

    public void setFifthGridRowCode(Integer fifthGridRowCode) {
        this.fifthGridRowCode = fifthGridRowCode;
    }

    public Integer getFifthGridColCode() {
        return fifthGridColCode;
    }

    public void setFifthGridColCode(Integer fifthGridColCode) {
        this.fifthGridColCode = fifthGridColCode;
    }

    public Integer getSixthGridRowCode() {
        return sixthGridRowCode;
    }

    public void setSixthGridRowCode(Integer sixthGridRowCode) {
        this.sixthGridRowCode = sixthGridRowCode;
    }

    public Integer getSixthGridColCode() {
        return sixthGridColCode;
    }

    public void setSixthGridColCode(Integer sixthGridColCode) {
        this.sixthGridColCode = sixthGridColCode;
    }

}

作者 east

Flink, Spark 4月 13,2022

Flink和Spark的Transformation不同地方对比

1、合并输入流：

在spark有Union
返回一个包含源DStream与其他 DStream的元素合并后的新DSTREAM。具体例子可以参考Spark Streaming多个输入流

在Flink中更高级，除了有union合并多个输入流（
union()所连接的两个或多个数据流的数据类型必须一致），还有connect（）。

①connect()只能连接两个数据流，union()可以连接多个数据流。

②connect()所连接的两个数据流的数据类型可以不一致，union()所连接的两个或多个数据流的数据类型必须一致。

③两个DataStream经过connect()之后被转化为ConnectedStreams，ConnectedStreams会对两个流的数据应用不同的处理方法，且两个流之间可以共享状态。

2、求最大最小的操作

Flink有 max()、 maxBy() 对该字段求最大值。
min()、minBy对某字段求最小值

作者 east

Hive 4月 10,2022

生产环境选型：Hive对比Pig

Apache Hadoop 包括一个不断增长的软件库，可帮助用户管理数据。对于需要管理大量信息的组织而言，Hive 和 Pig 是两个最关键的 Hadoop 项目。以下 Hive 与 Pig 比较将帮助您确定哪个 Hadoop 组件更符合您的需求。您还将有机会了解替代 ETL 解决方案的优势，这些解决方案使数据管理和丰富变得更加容易。

Hive vs Pig：最关键的区别
显然，Hive 和 Pig 为用户提供了很多优势。您使用的工具可能取决于您的数据需求。你是数据分析师还是程序员？您使用结构化数据还是半结构化数据？

了解这些问题的答案将帮助您确定更适合您的选项。通过了解 Hive 与 Pig 最关键的区别，您可以专注于适合您和您的组织的工具。

Hive 具有将数据转换为报告的可靠功能，而 Pig 为您提供了一种编程语言，可帮助您从一个或多个数据库中提取所需的信息。
Hive 在服务器端工作，而 Pig 在集群的客户端工作。
Hive 可以访问原始数据，而 Pig Latin 脚本不能。
HiveQL 遵循数据分析师可以轻松掌握的声明性 SQL 语言，而 Pig 依赖于具有更大学习曲线的 SQL 变体。
Hive 处理结构化数据，而 Pig 可以处理结构化和半结构化数据。

什么是 Hive Hadoop？
Apache 的 Hadoop Hive 组件执行多种功能，帮助数据分析专业人员通过类似于 SQL 的操作界面定位和丰富数据。如果您的团队成员已经了解 SQL，那么他们很容易开始使用 Hive。

数据分析师经常使用 Hive 来：

分析数据。
查询大量非结构化数据。
生成数据摘要。
Hive 为您提供了一种可靠的方式来定位和分析非结构化数据。显然，Hive 并不是每个组织的完美工具，但它具有出色的功能，使其成为需要有效方式处理非结构化数据的团体的有用工具。

什么是Pig Hadoop？
Apache Pig 使用脚本语言 Pig Latin 从 Hadoop 中查找、提取和丰富数据结构化和半结构化数据。许多人发现 Pig Latin 有点难学。但是，克服学习曲线可以让用户更好地控制他们的 Hadoop 数据。

选择Pig的人经常指出它：

快速加载数据。
隐式定义表模式。
支持同组。
像所有数据工具一样，Pig 也有其优点和缺点。您可以深入了解以下优点和缺点，以帮助您确定是否要将 Pig 作为 Hadoop 策略的一部分。

Apache Hadoop 在 ETL 中的作用
有些人错误地认为 Apache Hadoop 是一种 ETL 工具，它为他们提供了提取、转换和加载数据所需的所有工具。 Hadoop 提供了一些出色的优势，但它不属于 ETL 类别。但是，如果使用得当，它可以改进 ETL 策略和项目。

许多人使用 Apache Hadoop 之类的数据，因为它可以：

提高性能并防止硬件出现故障。
在将流行类型的数据移动到 ETL 管道之前集成它们。
提高处理和传输大数据的速度。
在用户将受损数据转移到其他工具之前识别安全漏洞并警告用户。
注意可能擦除或损坏数据的风险，让您有机会在丢失项目的关键信息之前解决问题。
虽然将 Hadoop 称为 ETL 解决方案是不正确的，但您可以将其称为 ETL 助手。该解决方案有几个很棒的功能，可以提高 ETL 项目的速度和准确性。

Hive：优点和缺点
要了解有关 Hive 优缺点的更多信息，直接从经常使用 Hadoop 组件的人那里获取信息是有意义的。 TrustRadius 评论家给 Apache Hive 打了 7.8 分（满分 10 分）。

用户从 Hive 获得的一些优势包括：

对已经熟悉 SQL 的任何人都适用的简单查询。
可在需要时从多个服务器寻求增援的可扩展性。
为数据分析生成临时查询的选项。
它处理长时间运行的查询的能力如何。
它能够连接各种关系数据库，包括 Postgres 和 MySQL。
使用 Java 和 Python 编写自定义函数的选项。
简化 Hadoop 体验，尤其是当没有技术背景的人参与数据项目时。
这就是潜在用户在选择 Hive 时应该考虑的一系列积极特征。不过，用户也有很多批评。例如，Hive 的许多批评包括：

缺乏对在线处理数据的支持。
无法支持子查询，
更新数据的复杂方法。
即席查询速度慢。
缺乏让管理员为用户分配特定角色的安全控制。
将易用性置于处理速度之上，尤其是在批处理方面。
尽管许多用户赞赏 Hive 的查询语言是基于 SQL 构建的，但他们指出 Hive 遗漏了一些非常有用的 SQL 命令。这种缺陷迫使用户浪费时间重写应该自动附带 Hadoop 组件的命令。

Pig：优点和缺点
Apache Pig 的数值审查略胜 Apache Hive。 TrustRadius 用户给 Pig 打了 7.9 分（满分 10 分）。

Apache Pig 用户提到的一些优点包括：

与 MapReduce、Spark 和 Tez 一起使用的快速执行。
它能够处理几乎任何数量的数据，无论大小。
使其与其他工具（如 Hive 和 DBMS）结合以改进其功能的功能。
一个强大的文档流程，可帮助新用户学习 Pig Latin。
本地和远程互操作性，让专业人员可以在任何地方通过可靠的连接工作。
尽管很多人喜欢 Apache Pig，但它确实存在给用户带来问题的问题。针对 Pig 的一些投诉集中在：

无法解决复杂的数学问题。
难以实施顺序检查。
很少有用于循环数据的选项，这会增加用户的工作量。
有些人难以掌握的特定领域语言（猪拉丁语）。
显然，Apache 可以对 Pig 进行一些改进。然而，它确实填补了一个吸引某些用户的利基空间。

作者 east

Flink 4月 8,2022

Flink面试题汇总

1、Flink如何保证精确一次性消费

Flink 保证精确一次性消费主要依赖于两种Flink机制

1、Checkpoint机制

2、二阶段提交机制

Checkpoint机制

主要是当Flink开启Checkpoint的时候，会往Source端插入一条barrir，然后这个barrir随着数据流向一直流动，当流入到一个算子的时候，这个算子就开始制作checkpoint，制作的是从barrir来到之前的时候当前算子的状态，将状态写入状态后端当中。然后将barrir往下流动，当流动到keyby 或者shuffle算子的时候，例如当一个算子的数据，依赖于多个流的时候，这个时候会有barrir对齐，也就是当所有的barrir都来到这个算子的时候进行制作checkpoint，依次进行流动，当流动到sink算子的时候，并且sink算子也制作完成checkpoint会向jobmanager 报告 checkpoint n 制作完成。

二阶段提交机制

Flink 提供了CheckpointedFunction与CheckpointListener这样两个接口，CheckpointedFunction中有snapshotState方法，每次checkpoint触发执行方法，通常会将缓存数据放入状态中，可以理解为一个hook，这个方法里面可以实现预提交，CheckpointListyener中有notifyCheckpointComplete方法，checkpoint完成之后的通知方法，这里可以做一些额外的操作。例如FLinkKafkaConumerBase使用这个来完成Kafka offset的提交，在这个方法里面可以实现提交操作。在2PC中提到如果对应流程例如某个checkpoint失败的话，那么checkpoint就会回滚，不会影响数据一致性，那么如果在通知checkpoint成功的之后失败了，那么就会在initalizeSate方法中完成事务的提交，这样可以保证数据的一致性。最主要是根据checkpoint的状态文件来判断的。

2、flink和spark区别

flink是一个类似spark的“开源技术栈”，因为它也提供了批处理，流式计算，图计算，交互式查询，机器学习等。flink也是内存计算，比较类似spark，但是不一样的是，spark的计算模型基于RDD，将流式计算看成是特殊的批处理，他的DStream其实还是RDD。而flink吧批处理当成是特殊的流式计算，但是批处理和流式计算的层的引擎是两个，抽象了DataSet和DataStream。flink在性能上也表现的很好，流式计算延迟比spark少，能做到真正的流式计算，而spark只能是准流式计算。而且在批处理上，当迭代次数变多，flink的速度比spark还要快，所以如果flink早一点出来，或许比现在的Spark更火。

3、Flink的状态可以用来做什么？

Flink状态主要有两种使用方式：

checkpoint的数据恢复
逻辑计算

4、Flink的waterMark机制，Flink watermark传递机制

Flink 中的watermark机制是用来处理乱序的，flink的时间必须是event time ，有一个简单的例子就是，假如窗口是5秒，watermark是2秒，那么总共就是7秒，这个时候什么时候会触发计算呢，假设数据初始时间是1000，那么等到6999的时候会触发5999窗口的计算，那么下一个就是13999的时候触发10999的窗口

其实这个就是watermark的机制，在多并行度中，例如在kafka中会所有的分区都达到才会触发窗口

5、Flink的时间语义

Event Time 事件产生的时间

Ingestion time 事件进入Flink的时间

processing time 事件进入算子的时间

6、Flink window join

1、window join，即按照指定的字段和滚动滑动窗口和会话窗口进行 inner join

2、是coGoup 其实就是left join 和 right join，

3、interval join 也就是在窗口中进行join 有一些问题，因为有些数据是真的会后到的，时间还很长，那么这个时候就有了interval join但是必须要是事件时间，并且还要指定watermark和水位以及获取事件时间戳。并且要设置偏移区间，因为join 也不能一直等的。

7、flink窗口函数有哪些

Tumbing window

Silding window

Session window

Count winodw

8、keyedProcessFunction 是如何工作的。假如是event time的话

keyedProcessFunction 是有一个ontime 操作的，假如是 event时间的时候那么调用的时间就是查看，event的watermark 是否大于 trigger time 的时间，如果大于则进行计算，不大于就等着，如果是kafka的话，那么默认是分区键最小的时间来进行触发。

9、flink是怎么处理离线数据的例如和离线数据的关联？

1、async io

2、broadcast

3、async io + cache

4、open方法中读取，然后定时线程刷新，缓存更新是先删除，之后再来一条之后再负责写入缓存

10、flink支持的数据类型

DataSet Api 和 DataStream Api、Table Api

11、Flink出现数据倾斜怎么办

Flink数据倾斜如何查看：

在flink的web ui中可以看到数据倾斜的情况，就是每个subtask处理的数据量差距很大，例如有的只有一M 有的100M 这就是严重的数据倾斜了。

KafkaSource端发生的数据倾斜

例如上游kafka发送的时候指定的key出现了数据热点问题，那么就在接入之后，做一个负载均衡（前提下游不是keyby）。

聚合类算子数据倾斜

预聚合加全局聚合

12、flink 维表关联怎么做的

1、async io

2、broadcast

3、async io + cache

4、open方法中读取，然后定时线程刷新，缓存更新是先删除，之后再来一条之后再负责写入缓存

13、Flink checkpoint的超时问题如何解决。

1、是否网络问题

2、是否是barrir问题

3、查看webui，是否有数据倾斜

4、有数据倾斜的话，那么解决数据倾斜后，会有改善，

14、flinkTopN与离线的TopN的区别

topn 无论是在离线还是在实时计算中都是比较常见的功能，不同于离线计算中的topn，实时数据是持续不断的，这样就给topn的计算带来很大的困难，因为要持续在内存中维持一个topn的数据结构，当有新数据来的时候，更新这个数据结构

15、sparkstreaming 和flink 里checkpoint的区别

sparkstreaming 的checkpoint会导致数据重复消费

但是flink的 checkpoint可以保证精确一次性，同时可以进行增量，快速的checkpoint的，有三个状态后端，memery、rocksdb、hdfs

16、简单介绍一下cep状态编程

Complex Event Processing（CEP）：

FLink Cep 是在FLink中实现的复杂时间处理库，CEP允许在无休止的时间流中检测事件模式，让我们有机会掌握数据中重要的部分，一个或多个由简单事件构成的时间流通过一定的规则匹配，然后输出用户想得到的数据，也就是满足规则的复杂事件。

17、 Flink cep连续事件的可选项有什么

18、如何通过flink的CEP来实现支付延迟提醒

19、Flink cep 你用过哪些业务场景

20、cep底层如何工作

21、cep怎么老化

22、cep性能调优

23、Flink的背压，介绍一下Flink的反压，你们是如何监控和发现的呢。

Flink 没有使用任何复杂的机制来解决反压问题，Flink 在数据传输过程中使用了分布式阻塞队列。我们知道在一个阻塞队列中，当队列满了以后发送者会被天然阻塞住，这种阻塞功能相当于给这个阻塞队列提供了反压的能力。

当你的任务出现反压时，如果你的上游是类似 Kafka 的消息系统，很明显的表现就是消费速度变慢，Kafka 消息出现堆积。

如果你的业务对数据延迟要求并不高，那么反压其实并没有很大的影响。但是对于规模很大的集群中的大作业，反压会造成严重的“并发症”。首先任务状态会变得很大，因为数据大规模堆积在系统中，这些暂时不被处理的数据同样会被放到“状态”中。另外，Flink 会因为数据堆积和处理速度变慢导致 checkpoint 超时，而 checkpoint 是 Flink 保证数据一致性的关键所在，最终会导致数据的不一致发生。

Flink Web UI

Flink 的后台页面是我们发现反压问题的第一选择。Flink 的后台页面可以直观、清晰地看到当前作业的运行状态。

Web UI，需要注意的是，只有用户在访问点击某一个作业时，才会触发反压状态的计算。在默认的设置下，Flink的TaskManager会每隔50ms触发一次反压状态监测，共监测100次，并将计算结果反馈给JobManager，最后由JobManager进行反压比例的计算，然后进行展示。

在生产环境中Flink任务有反压有三种OK、LOW、HIGH

OK正常

LOW一般

HIGH高负载

24、Flink的CBO，逻辑执行计划和物理执行计划

Flink的优化执行其实是借鉴的数据库的优化器来生成的执行计划。

CBO，成本优化器，代价最小的执行计划就是最好的执行计划。传统的数据库，成本优化器做出最优化的执行计划是依据统计信息来计算的。Flink 的成本优化器也一样。Flink 在提供最终执行前，优化每个查询的执行逻辑和物理执行计划。这些优化工作是交给底层来完成的。根据查询成本执行进一步的优化，从而产生潜在的不同决策：如何排序连接，执行哪种类型的连接，并行度等等。

// TODO

25、Flink中数据聚合，不使用窗口怎么实现聚合

valueState 用于保存单个值
ListState 用于保存list元素
MapState 用于保存一组键值对
ReducingState 提供了和ListState相同的方法，返回一个ReducingFunction聚合后的值。
AggregatingState和 ReducingState类似，返回一个AggregatingState内部聚合后的值

26、Flink中state有哪几种存储方式

Memery、RocksDB、HDFS

27、Flink 异常数据怎么处理

异常数据在我们的场景中，一般分为缺失字段和异常值数据。

异常值： 例如宝宝的年龄的数据，例如对于母婴行业来讲，一个宝宝的年龄是一个至关重要的数据，可以说是最重要的，因为宝宝大于3岁几乎就不会在母婴上面购买物品。像我们的有当日、未知、以及很久的时间。这样都属于异常字段，这些数据我们会展示出来给店长和区域经理看，让他们知道多少个年龄是不准的。如果要处理的话，可以根据他购买的时间来进行实时矫正，例如孕妇服装、奶粉的段位、纸尿裤的大小，以及奶嘴啊一些能够区分年龄段的来进行处理。我们并没有实时处理这些数据，我们会有一个底层的策略任务夜维去跑，一个星期跑一次。

缺失字段： 例如有的字段真的缺失的很厉害，能修补就修补。不能修补就放弃，就像上家公司中的新闻推荐过滤器。

28、Flink 监控你们怎么做的

1、我们监控了Flink的任务是否停止

2、我们监控了Flink的Kafka的LAG

3、我们会进行实时数据对账，例如销售额。

29、Flink 有数据丢失的可能吗

Flink有三种数据消费语义：

At Most Once 最多消费一次发生故障有可能丢失
At Least Once 最少一次发生故障有可能重复
Exactly-Once 精确一次如果产生故障，也能保证数据不丢失不重复。

flink 新版本已经不提供 At-Most-Once 语义。

30、Flink interval join 你能简单的写一写吗

DataStream<T> keyed1 = ds1.keyBy(o -> o.getString("key"))
DataStream<T> keyed2 = ds2.keyBy(o -> o.getString("key"))
//右边时间戳-5s<=左边流时间戳<=右边时间戳-1s
keyed1.intervalJoin(keyed2).between(Time.milliseconds(-5), Time.milliseconds(5))

31、Flink 提交的时候并行度如何制定，以及资源如何配置

并行度根据kafka topic的并行度，一个并行度3个G

32、Flink的boardcast join 的原理是什么

利用 broadcast State 将维度数据流广播到下游所有 task 中。这个 broadcast 的流可以与我们的事件流进行 connect，然后在后续的 process 算子中进行关联操作即可。

33、flink的source端断了，比如kafka出故障，没有数据发过来，怎么处理？

会有报警，监控的kafka偏移量也就是LAG。

34、flink有什么常用的流的API?

window join 啊 cogroup 啊 map flatmap，async io 等

35、flink的水位线，你了解吗，能简单介绍一下吗

Flink 的watermark是一种延迟触发的机制。

一般watermark是和window结合来进行处理乱序数据的，Watermark最根本就是一个时间机制，例如我设置最大乱序时间为2s，窗口时间为5秒，那么就是当事件时间大于7s的时候会触发窗口。当然假如有数据分区的情况下，例如kafka中接入watermake的话，那么watermake是会流动的，取的是所有分区中最小的watermake进行流动，因为只有最小的能够保证，之前的数据都已经来到了，可以触发计算了。

36、Flink怎么维护Checkpoint？在HDFS上存储的话会有小文件吗

默认情况下，如果设置了Checkpoint选项，Flink只保留最近成功生成的1个Checkpoint。当Flink程序失败时，可以从最近的这个Checkpoint来进行恢复。但是，如果我们希望保留多个Checkpoint，并能够根据实际需要选择其中一个进行恢复，这样会更加灵活。Flink支持保留多个Checkpoint，需要在Flink的配置文件conf/flink-conf.yaml中，添加如下配置指定最多需要保存Checkpoint的个数。

关于小文件问题可以参考代达罗斯之殇-大数据领域小文件问题解决攻略。

37、Spark和Flink的序列化，有什么区别吗？

Spark 默认使用的是 Java序列化机制，同时还有优化的机制，也就是kryo

Flink是自己实现的序列化机制，也就是TypeInformation

38、Flink是怎么处理迟到数据的？但是实际开发中不能有数据迟到，怎么做？

Flink 的watermark是一种延迟触发的机制。

39、画出flink执行时的流程图。

40、Flink分区分配策略

41、Flink关闭后状态端数据恢复得慢怎么办？

42、了解flink的savepoint吗？讲一下savepoint和checkpoint的不同和各有什么优势

43、flink的状态后端机制

Flink的状态后端是Flink在做checkpoint的时候将状态快照持久化，有三种状态后端 Memery、HDFS、RocksDB

44、flink中滑动窗口和滚动窗口的区别，实际应用的窗口是哪种？用的是窗口长度和滑动步长是多少？

45、用flink能替代spark的批处理功能吗

Flink 未来的目标是批处理和流处理一体化，因为批处理的数据集你可以理解为是一个有限的数据流。Flink 在批出理方面，尤其是在今年 Flink 1.9 Release 之后，合入大量在 Hive 方面的功能，你可以使用 Flink SQL 来读取 Hive 中的元数据和数据集，并且使用 Flink SQL 对其进行逻辑加工，不过目前 Flink 在批处理方面的性能，还是干不过 Spark的。

目前看来，Flink 在批处理方面还有很多内容要做，当然，如果是实时计算引擎的引入，Flink 当然是首选。

46、flink计算的UV你们是如何设置状态后端保存数据

可以使用布隆过滤器。

47、sparkstreaming和flink在执行任务上有啥区别，不是简单的流处理和微批，sparkstreaming提交任务是分解成stage，flink是转换graph，有啥区别？

48、flink把streamgraph转化成jobGraph是在哪个阶段？

49、Flink中的watermark除了处理乱序数据还有其他作用吗？

还有kafka数据顺序消费的处理。

50、flink你一般设置水位线设置多少

我们之前设置的水位线是6s

52、Flink任务提交流程

Flink任务提交后，Client向HDFS上传Flink的jar包和配置，之后向Yarn ResourceManager提交任务，ResourceManager分配Container资源并通知对应的NodeManager启动
ApplicationMaster，ApplicationMaster启动后加载Flink的jar包和配置构建环境，然后启动JobManager；之后Application Master向ResourceManager申请资源启动TaskManager
，ResourceManager分配Container资源后，由ApplicationMaster通知资源所在的节点的NodeManager启动TaskManager，NodeManager加载Flink的Jar包和配置构建环境并启动TaskManager，TaskManager启动向JobManager发送心跳，并等待JobManager向其分配任务。

53、Flink技术架构图

54、flink如何实现在指定时间进行计算。

55、手写Flink topN

57、Flink的Join算子有哪些

一般join是发生在window上面的:

1、window join，即按照指定的字段和滚动滑动窗口和会话窗口进行 inner join

2、是coGoup 其实就是left join 和 right join，

58、Flink1.10 有什么新特性吗？

内存管理及配置优化

Flink 目前的 TaskExecutor 内存模型存在着一些缺陷，导致优化资源利用率比较困难，例如：

流和批处理内存占用的配置模型不同
流处理中的 RocksDB state backend 需要依赖用户进行复杂的配置

为了让内存配置变的对于用户更加清晰、直观，Flink 1.10 对 TaskExecutor 的内存模型和配置逻辑进行了较大的改动（FLIP-49 [7]）。这些改动使得 Flink 能够更好地适配所有部署环境（例如 Kubernetes, Yarn, Mesos），让用户能够更加严格的控制其内存开销。

Managed 内存扩展

Managed 内存的范围有所扩展，还涵盖了 RocksDB state backend 使用的内存。尽管批处理作业既可以使用堆内内存也可以使用堆外内存，使用 RocksDB state backend 的流处理作业却只能利用堆外内存。因此为了让用户执行流和批处理作业时无需更改集群的配置，我们规定从现在起 managed 内存只能在堆外。

简化 RocksDB 配置

此前，配置像 RocksDB 这样的堆外 state backend 需要进行大量的手动调试，例如减小 JVM 堆空间、设置 Flink 使用堆外内存等。现在，Flink 的开箱配置即可支持这一切，且只需要简单地改变 managed 内存的大小即可调整 RocksDB state backend 的内存预算。

另一个重要的优化是，Flink 现在可以限制 RocksDB 的 native 内存占用，以避免超过总的内存预算—这对于 Kubernetes 等容器化部署环境尤为重要。

统一的作业提交逻辑
在此之前，提交作业是由执行环境负责的，且与不同的部署目标（例如 Yarn, Kubernetes, Mesos）紧密相关。这导致用户需要针对不同环境保留多套配置，增加了管理的成本。

在 Flink 1.10 中，作业提交逻辑被抽象到了通用的 Executor 接口。新增加的 ExecutorCLI （引入了为任意执行目标指定配置参数的统一方法。此外，随着引入 JobClient负责获取 JobExecutionResult，获取作业执行结果的逻辑也得以与作业提交解耦。

原生 Kubernetes 集成（Beta）

对于想要在容器化环境中尝试 Flink 的用户来说，想要在 Kubernetes 上部署和管理一个 Flink standalone 集群，首先需要对容器、算子及像 kubectl 这样的环境工具有所了解。

在 Flink 1.10 中，我们推出了初步的支持 session 模式的主动 Kubernetes 集成（FLINK-9953）。其中，“主动”指 Flink ResourceManager (K8sResMngr) 原生地与 Kubernetes 通信，像 Flink 在 Yarn 和 Mesos 上一样按需申请 pod。用户可以利用 namespace，在多租户环境中以较少的资源开销启动 Flink。这需要用户提前配置好 RBAC 角色和有足够权限的服务账号。

Table API/SQL: 生产可用的 Hive 集成

Flink 1.9 推出了预览版的 Hive 集成。该版本允许用户使用 SQL DDL 将 Flink 特有的元数据持久化到 Hive Metastore、调用 Hive 中定义的 UDF 以及读、写 Hive 中的表。Flink 1.10 进一步开发和完善了这一特性，带来了全面兼容 Hive 主要版本的生产可用的 Hive 集成。

Batch SQL 原生分区支持

此前，Flink 只支持写入未分区的 Hive 表。在 Flink 1.10 中，Flink SQL 扩展支持了 INSERT OVERWRITE 和 PARTITION 的语法（FLIP-63 ），允许用户写入 Hive 中的静态和动态分区。

写入静态分区

INSERT { INTO | OVERWRITE } TABLE tablename1 [PARTITION (partcol1=val1, partcol2=val2 ...)] select_statement1 FROM from_statement;

写入动态分区

INSERT { INTO | OVERWRITE } TABLE tablename1 select_statement1 FROM from_statement;

对分区表的全面支持，使得用户在读取数据时能够受益于分区剪枝，减少了需要扫描的数据量，从而大幅提升了这些操作的性能。

另外，除了分区剪枝，Flink 1.10 的 Hive 集成还引入了许多数据读取方面的优化，例如：

投影下推：Flink 采用了投影下推技术，通过在扫描表时忽略不必要的域，最小化 Flink 和 Hive 表之间的数据传输量。这一优化在表的列数较多时尤为有效。
LIMIT 下推：对于包含 LIMIT 语句的查询，Flink 在所有可能的地方限制返回的数据条数，以降低通过网络传输的数据量。
读取数据时的 ORC 向量化：为了提高读取 ORC 文件的性能，对于 Hive 2.0.0 及以上版本以及非复合数据类型的列，Flink 现在默认使用原生的 ORC 向量化读取器。

59、Flink的重启策略

固定延迟重启策略

固定延迟重启策略是尝试给定次数重新启动作业。如果超过最大尝试次数，则作业失败。在两次连续重启尝试之间，会有一个固定的延迟等待时间。

故障率重启策略

故障率重启策略在故障后重新作业，当设置的故障率（failure rate）超过每个时间间隔的故障时，作业最终失败。在两次连续重启尝试之间，重启策略延迟等待一段时间。

无重启策略

作业直接失败，不尝试重启。

后备重启策略

使用群集定义的重新启动策略。这对于启用检查点的流式传输程序很有帮助。默认情况下，如果没有定义其他重启策略，则选择固定延迟重启策略。

60、Flink什么时候用aggregate()或者process()

aggregate： 增量聚合

process： 全量聚合

当计算累加操作时候可以使用aggregate操作。

当计算窗口内全量数据的时候使用process，例如排序等操作。

61、Flink优化你了解多少

62、Flink内存溢出怎么办

63、说说Flink中的keyState包含哪些数据结构

64、Flink shardGroup的概念

作者 east

Java 4月 7,2022

地图不同坐标系的转换工具类

提供以下的坐标系互相转换：

GCJ-02(火星坐标系)转换为WGS84坐标系

BD-09(百度坐标系)转换为WGS84坐标系

BD-09(百度坐标系)转换为GCJ-02(火星坐标)

sogou转换为WGS84

/**
 * @author jy 地图转换测试/
 */
public final class MapUtil {

    /**
     * 辅助类禁止共有构造函数
     */
    private MapUtil() {

    }

    /**
     * 常量:pi
     */
    public static final double CONSTANT_PI = 3.1415926535897932384626;

    /**
     * 常量:a
     */
    public static final double CONSTANT_A = 6378245.0;

    /**
     * 常量:e
     */
    public static final double CONSTANT_E = 0.00669342162296594323;

    /**
     * 常量:ccc
     */
    private static final double[] CCC = new double[]{12890594.86, 8362377.87, 5591021, 3481989.83, 1678043.12, 0};

    /**
     * 常量:ddd
     */
    private static final double[][] DDD = {
            new double[]{1.410526172116255e-8, 0.00000898305509648872, -1.9939833816331, 200.9824383106796,
                    -187.2403703815547, 91.6087516669843, -23.38765649603339, 2.57121317296198, -0.03801003308653,
                    17337981.2},
            new double[]{-7.435856389565537e-9, 0.000008983055097726239, -0.78625201886289, 96.32687599759846,
                    -1.85204757529826, -59.36935905485877, 47.40033549296737, -16.50741931063887, 2.28786674699375,
                    10260144.86},
            new double[]{-3.030883460898826e-8, 0.00000898305509983578, 0.30071316287616, 59.74293618442277,
                    7.357984074871, -25.38371002664745, 13.45380521110908, -3.29883767235584, 0.32710905363475,
                    6856817.37},
            new double[]{-1.981981304930552e-8, 0.000008983055099779535, 0.03278182852591, 40.31678527705744,
                    0.65659298677277, -4.44255534477492, 0.85341911805263, 0.12923347998204, -0.04625736007561,
                    4482777.06},
            new double[]{3.09191371068437e-9, 0.000008983055096812155, 0.00006995724062, 23.10934304144901,
                    -0.00023663490511, -0.6321817810242, -0.00663494467273, 0.03430082397953, -0.00466043876332,
                    2555164.4},
            new double[]{2.890871144776878e-9, 0.000008983055095805407, -3.068298e-8, 7.47137025468032,
                    -0.00000353937994, -0.02145144861037, -0.00001234426596, 0.00010322952773, -0.00000323890364,
                    826088.5}};

    /**
     * GCJ-02(火星坐标系)转换为WGS84坐标系<br>
     * GCJ-02:谷歌中国地图、搜搜中国地图、高德地图<br>
     *
     * @param lat 纬度
     * @param lon 经度
     * @return
     */
    public static Coord gcj02ToWgs84(double lat, double lon) {

        Coord coord = transform(lat, lon);

        double finalLon = lon * 2 - coord.getLon();
        double finalLat = lat * 2 - coord.getLat();

        return new Coord(finalLat, finalLon);

    }

    /**
     * BD-09(百度坐标系)转换为WGS84坐标系<br>
     *
     * @param lat 纬度
     * @param lon 经度
     * @return
     */
    public static Coord bd09ToWgs84(double lat, double lon) {

        Coord gcj02 = MapUtil.bd09ToGcj02(lat, lon);
        Coord wgs84 = MapUtil.gcj02ToWgs84(gcj02.getLat(), gcj02.getLon());

        return wgs84;
    }

    /**
     * BD-09(百度坐标系)转换为GCJ-02(火星坐标)<br>
     *
     * @param lat 纬度
     * @param lon 经度
     * @return
     */
    public static Coord bd09ToGcj02(double lat, double lon) {

        double x = lon - 0.0065;
        double y = lat - 0.006;
        double z = Math.sqrt(x * x + y * y) - 0.00002 * Math.sin(y * CONSTANT_PI);

        double theta = Math.atan2(y, x) - 0.000003 * Math.cos(x * CONSTANT_PI);

        double finalLon = z * Math.cos(theta);
        double finalLat = z * Math.sin(theta);

        return new Coord(finalLat, finalLon);

    }

    /**
     * sogou转换为WGS84
     *
     * @param lat 纬度
     * @param lon 经度
     * @return
     */
    public static Coord sogouToWgs84(double lat, double lon) {

        Coord coord = new Coord(lat, lon);

        Coord tmp = new Coord();
        tmp.setLon(Math.abs(coord.getLon()));
        tmp.setLat(Math.abs(coord.getLat()));

        double[] obj = new double[9];
        for (int i = 0; i < CCC.length; i++) {
            if (tmp.getLat() > CCC[i]) {
                obj = DDD[i];
                break;
            }
        }

        return MapUtil.convert(coord, obj);
    }

    /**
     * sogou内部转换函数
     *
     * @param coord
     * @param varAttr
     * @return
     */
    private static Coord convert(Coord coord, double[] varAttr) {
        Coord tmp = new Coord();
        tmp.setLon(varAttr[0] + varAttr[1] * Math.abs(coord.getLon()));
        double f = Math.abs(coord.getLat()) / varAttr[9];
        double tmpLat = varAttr[2] + varAttr[3] * f + varAttr[4] * f * f + varAttr[5] * f * f * f
                + varAttr[6] * f * f * f * f + varAttr[7] * f * f * f * f * f + varAttr[8] * f * f * f * f * f * f;
        tmp.setLat(tmpLat);
        tmp.setLon(tmp.getLon() * (coord.getLon() < 0 ? -1 : 1));
        tmp.setLat(tmp.getLat() * (coord.getLat() < 0 ? -1 : 1));

        return tmp;
    }

    /**
     * 坐标点转换函数
     *
     * @param lat 纬度
     * @param lon 经度
     * @return
     */
    private static Coord transform(double lat, double lon) {

        if (outOfChina(lat, lon)) {
            return new Coord(lat, lon);
        }

        double dLat = transformLat(lat - 35.0, lon - 105.0);
        double dLon = transformLon(lat - 35.0, lon - 105.0);

        double radLat = lat / 180.0 * CONSTANT_PI;

        double magic = Math.sin(radLat);
        magic = 1 - CONSTANT_E * magic * magic;
        double sqrtMagic = Math.sqrt(magic);
        dLat = (dLat * 180.0) / ((CONSTANT_A * (1 - CONSTANT_E)) / (magic * sqrtMagic) * CONSTANT_PI);
        dLon = (dLon * 180.0) / (CONSTANT_A / sqrtMagic * Math.cos(radLat) * CONSTANT_PI);
        double mgLat = lat + dLat;
        double mgLon = lon + dLon;

        return new Coord(mgLat, mgLon);

    }

    /**
     * 纬度转换函数
     *
     * @param lat 维度
     * @param lon 经度
     * @return
     */
    private static double transformLat(double lat, double lon) {

        double ret = -100.0 + 2.0 * lon + 3.0 * lat + 0.2 * lat * lat + 0.1 * lon * lat
                + 0.2 * Math.sqrt(Math.abs(lon));
        ret += (20.0 * Math.sin(6.0 * lon * CONSTANT_PI) + 20.0 * Math.sin(2.0 * lon * CONSTANT_PI)) * 2.0 / 3.0;
        ret += (20.0 * Math.sin(lat * CONSTANT_PI) + 40.0 * Math.sin(lat / 3.0 * CONSTANT_PI)) * 2.0 / 3.0;
        ret += (160.0 * Math.sin(lat / 12.0 * CONSTANT_PI) + 320 * Math.sin(lat * CONSTANT_PI / 30.0)) * 2.0 / 3.0;

        return ret;
    }

    /**
     * 经度转换函数
     *
     * @param lat 纬度
     * @param lon 经度
     * @return
     */
    private static double transformLon(double lat, double lon) {

        double ret = 300.0 + lon + 2.0 * lat + 0.1 * lon * lon + 0.1 * lon * lat + 0.1 * Math.sqrt(Math.abs(lon));

        ret += (20.0 * Math.sin(6.0 * lon * CONSTANT_PI) + 20.0 * Math.sin(2.0 * lon * CONSTANT_PI)) * 2.0 / 3.0;
        ret += (20.0 * Math.sin(lon * CONSTANT_PI) + 40.0 * Math.sin(lon / 3.0 * CONSTANT_PI)) * 2.0 / 3.0;
        ret += (150.0 * Math.sin(lon / 12.0 * CONSTANT_PI) + 300.0 * Math.sin(lon / 30.0 * CONSTANT_PI)) * 2.0 / 3.0;

        return ret;
    }

    /**
     * 判断坐标点是否在中国范围内
     *
     * @param lat
     * @param lon
     * @return
     */
    private static boolean outOfChina(double lat, double lon) {

        if (lon < 72.004 || lon > 137.8347) {
            return true;
        }

        if (lat < 0.8293 || lat > 55.8271) {
            return true;
        }

        return false;
    }

}

public class Coord implements Serializable {

    /**
     *
     */
    private static final long serialVersionUID = -229841606158709053L;
    /**
     * 纬度
     */
    private double lat;
    /**
     * 经度
     */
    private double lon;

    /**
     * 默认构造参数
     */
    public Coord() {

    }

    /**
     * 重载构造方法
     *
     * @param lat 纬度
     * @param lon 经度
     */
    public Coord(double lat, double lon) {
        this.lat = lat;
        this.lon = lon;
    }

    public double getLat() {
        return lat;
    }

    public void setLat(double lat) {
        this.lat = lat;
    }

    public double getLon() {
        return lon;
    }

    public void setLon(double lon) {
        this.lon = lon;
    }

}

作者 east

Hive, 数据库 4月 4,2022

Hive对比SQL：哪个更适合数据分析

Hive 和 SQL 之间的主要区别：

架构：Hive 是一个用于数据分析的数据仓库项目； SQL 是一种编程语言。（但是，Hive 通过称为 HiveQL 的编程语言执行数据分析，类似于 SQL。）

设置：Hive 是一个基于开源软件程序 Hadoop 的数据仓库。
数据分析：Hive 比 SQL 更有效地处理复杂数据，适合不太复杂的数据集。

价格：Hive 价格从每位用户每月 12 美元起。 SQL 是开源和免费的。

评论：Hive 在 G2 网站上的客户评论评分为 4.2/5。因为 SQL 是一种编程语言而不是“产品”，所以它在 G2 上没有评论。

大数据需要强大的工具。成功的组织查询、管理和分析来自数百个数据源的数千个数据集。这就是 Hive 和 SQL 等工具的用武之地。尽管非常不同，但查询和编程大数据都是如此。

但是哪种工具适合您的组织？在这篇评论中，我们在功能、价格、支持、用户评分等方面比较了 Hive 与 SQL。

什么是Hive？
Apache Hive 是一个用于数据查询和分析的数据仓库项目。 Hive 建立在 Apache Hadoop（一个用于处理大数据的开源程序）之上，通过查询语言 HiveQL 执行数据分析，它允许用户构建数据并生成各种有用的分析。

Hive 由 Facebook 开发，使希望从电子表格、网络日志、CRM 系统等中查询和汇总数据的用户受益。它在 Hadoop 分布式文件系统 (HDFS) 中查询数据，并将该系统用于自己的存储。它还运行 MapReduce 作业。

什么是 SQL？
结构化查询语言 (SQL) 是一种特定于领域的编程语言，用于管理数据和/或处理数据流。它主要管理数据并处理关系数据库管理系统中保存的实时数据。在这篇评论的上下文中，SQL 就像 HiveQL。

SQL 由 Oracle 开发，是一种用于分析查询的声明性语言。它比 Hive（和 HiveQL）更新得多。 SQL 可以追溯到 45 年前，并且在许多 IT 系统中已经无处不在。

有关我们的原生 SQL 连接器的更多信息，请访问我们的集成页面。

Hive 和 SQL 的区别
Hive 在 HDFS 中写入和查询数据。 SQL 需要多次读取和写入。
Hive 更适合分析复杂的数据集。 SQL 更适合快速分析不太复杂的数据集。
SQL 支持联机事务处理 (OLTP)。 Hive 不支持 OLTP。
Hive 查询可能具有高延迟，因为 Hive 通过 Hadoop 运行批处理。这意味着某些查询要等待一个小时（或更长时间）。更新 Hive 上的数据也可能需要很长时间。

支持和培训
HIVE
一个在线社区（Apache 软件基金会）
资源
邮件列表
语言手册

SQL
虽然没有提供官方培训，但有各种 SQL 第三方培训模块/支持社区。

价钱
HIVE
计划从每位用户每月 12 美元起。
有 14 天的免费试用期。
SQL
作为一个开源平台，SQL 是 100% 免费的。但是，SQL 定价并未考虑您可能遇到的任何设置或维护成本。

结论
Hive 和 SQL 是处理（和驯服！）大数据的两种工具。尽管这些工具有相似之处，但它们的差异足以保证进行比较。我们认为 Hive 更适合分析复杂的数据集，而 SQL 更适用于不太复杂的数据集，并且在执行这些任务时速度更快。另外，它是开源和免费的。最终，适合您的工具取决于您如何分析组织中的大数据。

作者 east

Hadoop 4月 4,2022

Hadoop对比SQL，哪个更适合数据管理

Hadoop 与 SQL 之间的主要区别：

架构：Hadoop 是一个开源框架（或“生态系统”），它在计算机/服务器集群之间分布数据集并并行处理数据。 SQL 是一种特定领域的编程语言，用于处理关系数据库中的数据。

数据：Hadoop 一次写入数据； SQL 多次写入数据。（Hadoop 和 SQL 多次读取数据。）

技能水平：Hadoop 比 SQL 更难学。（但是，两者都需要代码知识。）

价格：Hadoop 和 SQL 是开源的并且可以免费使用。但是，两者都会产生额外的设置和维护成本。

评论：Hadoop 在软件评论网站 G2.com 上的客户评分为 4.3/5。因为 SQL 是一种编程语言，而不是作为“产品”提供，所以它在 G2 上没有得分。

组织依靠大数据为其业务提供动力，但许多团队都在为数据管理的复杂性而苦苦挣扎。值得庆幸的是，Hadoop 和 SQL 更有效地处理大型数据集。这些工具以独特的方式管理数据，这使得我们很难在同类的基础上比较它们。但是，希望简化其技术堆栈的组织可能有理由选择其中一个。

在本文中，我们根据几个因素比较了 Hadoop 与 SQL，包括功能和客户评论分数。

什么是 Hadoop？
Apache Hadoop 是一个开源工具生态系统，可将数据集存储在分布式系统中并解决各种数据管理问题。

Hadoop 由四个组件组成：MapReduce、Yarn、库，以及最终在现成硬件上运行的 Hadoop 分布式文件系统 (HDFS)。 Hadoop 处理各种数据集，使其成为希望从大量来源生成有价值数据洞察的组织的绝佳选择。它有利于处理大量数据。

Hadoop 对跨计算机和服务器集群的数据集进行分布式处理。它以并行方式处理数据，因此它可以同时在多台机器上工作。 HDFS 存储提交的数据，MapReduce 处理数据，Yarn 划分数据管理任务。

世界上一些最成功的技术组织都使用 Hadoop，包括 IBM、Pivotal Software、Hadapt 和 Amazon Web Services。

什么是 SQL？
结构化查询语言 (SQL) 是一种开源的特定于领域的编程语言，用于在 Oracle、SQL Server 或 MySQL 等关系数据库管理系统 (RDMS) 中进行数据管理和处理数据流。 SQL 由 Oracle 开发，是一种用于分析查询的声明性语言。

有关我们的原生 SQL 连接器的更多信息，请访问我们的集成页面。

Hadoop 与 SQL：有什么区别？
也许 Hadoop 和 SQL 之间最大的区别在于这些工具管理和集成数据的方式。 SQL 只能处理有限的数据集，例如关系数据，并且难以处理更复杂的数据集。 Hadoop 可以处理大型数据集和非结构化数据。

当然，还有很多其他的区别：

Hadoop 线性扩展； SQL 是非线性的。
Hadoop的完整性低； SQL 是高完整性的。
Hadoop只能写一次； SQL 多次写入。
Hadoop具有动态模式结构； SQL 具有静态模式结构。
Hadoop 支持批处理（通过 HDFS）； SQL 没有。
Hadoop 比 SQL 更难学习，但更容易扩展。您可以轻松地将数据节点添加到 Hadoop 集群。
您选择的工具取决于您要管理的数据集。如果您需要处理大量数据，请选择 Hadoop。如果您不想要高级数据管理的复杂性，请选择 SQL。

作者 east

Spark 4月 3,2022

生产环境选型考虑：Spark和Tez有什么不同

让我们开始这场精彩的讨论。首先，退一步；我们已经指出 Apache Spark 和 Hadoop MapReduce 是两种不同的大数据利器。前者是高性能的内存数据处理框架，后者是成熟的PB级批处理平台。我们也知道 Apache Hive 和 HBase 是两个功能相似的非常不同的工具。 Hive 是运行 MapReduce 作业的类似 SQL 的引擎，而 HBase 是 Hadoop 上的 NoSQL 键/值数据库。

在纸面上，它们有很多共同点。两者都具有内存功能，都可以在 Hadoop YARN 之上运行，并且都支持来自任何数据源的所有数据类型。那么两者有什么区别呢？

Tez 非常适合 YARN 架构。 Spark 可能会遇到资源管理问题。

Spark 更适合主流开发人员，而 Tez 是专用工具的框架。

Spark 不能与 YARN 应用程序同时运行（目前）。 Tez 是专门为在 YARN 之上执行而构建的。

Tez 的容器可以在完成后关闭以节省资源。即使不处理数据，Spark 的容器也会占用资源。

这些只是高层次上的一些差异。在这里，我们将探索这些项目中的每一个。

什么是 Apache Spark？

Apache Spark 是一个用于处理大数据的开源分析引擎和集群计算框架。它是非营利性 Apache 软件基金会的创意，该基金会是一个致力于各种开源软件项目的去中心化组织。

它于 2014 年首次发布，基于 Hadoop MapReduce 分布式计算框架构建。它保留了 MapReduce 的许多优点——例如可扩展性和容错性——同时还提高了速度和易用性。

除了核心数据处理引擎，它还包括 SQL、机器学习和流处理库。该框架与 Java、Scala、Python 和 R 编程语言兼容，赢得了开发人员的广泛关注。它还支持第三方技术，如 Amazon S3、Hadoop 的 HDFS、MapR XD 和 NoSQL 数据库，如 Cassandra 和 MongoDB。

它的吸引力在于它能够将不同的流程、技术和技术整合到一个单一的大数据管道中，从而提高生产力和效率。由于其灵活性，它已成为大数据处理领域非常流行和有效的“瑞士军刀”。

什么是 Apache Tez？

Apache Tez 是一个基于 MapReduce 技术的大数据处理开源框架。两者都提供了一个执行引擎，可以使用有向无环图 (DAG) 来处理大量数据。

它通过将计算视为 DAG 来概括 MapReduce 范式。 MapReduce 任务组合成一个作业，该作业被视为 DAG 中的一个节点，执行并发和序列化。

同时，DAG 的边缘表示作业之间的数据移动。 Tez 与数据类型无关，因此它只关心数据的移动（而不是它采用的格式）。

通过改进 MapReduce 的一些限制，Tez 试图提高数据处理作业的性能。这种增加的效率使程序员能够做出他们认为最适合他们的项目的设计和开发选择。

Apache Spark 将自己标榜为“用于大规模数据处理的统一分析引擎”。同时，Apache Tez 称自己为“一个应用程序框架，它允许使用复杂的有向无环图来处理数据的任务”。

因为 Spark 也使用有向无环图，这两个工具听起来是不是很相似？可能是。但也有一些重要的区别需要考虑。以下是两者之间的根本区别：

差异#1：Hive和Pig

差异 #2：Hadoop YARN

差异#3：性能测试

我们将在下面的部分中详细介绍这些差异中的每一个。

他们支持Pig和Hive吗？

Hive 和 Pig 是两个用于大数据的开源 Apache 软件应用程序。 Hive 是一个数据仓库，而 Pig 是一个用于创建在 Hadoop 上运行的数据处理作业的平台。虽然两者都声称支持 Pig 和 Hive，但现实并不那么清楚。我们尝试使用 Spork 项目在 Spark 上运行 Pig，但遇到了一些问题；至少，在 Spark 上使用 Pig 充其量仍是不确定的。

使用YARN

YARN 是 Hadoop 的资源管理器和作业调度器。理论上，Spark 既可以作为独立应用程序执行，也可以在 YARN 之上执行。然而，Tez 是专门为在 YARN 之上执行而构建的。不过，Spark 不能与其他 YARN 应用程序同时运行（至少现在还不能）。

Tez 项目的开发人员之一 Gopal V 写了一篇关于他为什么喜欢 Tez 的详细文章。他的结论是：

“在我使用过的框架之间，这是 Tez 真正的区别特性——Tez 不需要容器保持运行来做任何事情，只需应用程序管理器在不同查询之间的空闲期间运行。您可以保留容器，但这是一种优化，而不是会话空闲期间的要求。”

他所说的“框架”也指 Spark——它的容器需要保持运行并占用资源，即使它们不处理任何数据。但是，Tez 容器可以在完成后立即关闭并释放资源。

大多数情况下，您无论如何都会使用基于 Hadoop 的应用程序，例如 Hive、HBase 甚至经典的 MapReduce。因此，您可以在任何 Hadoop 集群上安装 Spark，但您可能会遇到资源管理问题。另一方面，Tez 可以非常适合您的 YARN 架构，包括资源管理。

Apache Spark 的亮点：图形处理

GraphX 是扩展 Spark RDD 的图计算引擎。术语“图”是指图论中的图，而不是用于商业计算的图。图论中使用的图捕获数据之间的交互和依赖关系。

GraphX 最初是加州大学伯克利分校的一个研究项目。该项目后来被捐赠给了 Apache 软件基金会和 Spark 项目。

GraphX 不同于其他图计算引擎，因为它将图分析和 ETL 统一在一个平台上。 GraphX 还可以分析非图形形式的数据。其内存计算能力使 GraphX 比其他图形处理引擎更快。

图处理的常用场景

社交网络分析 – 用于识别影响者以进行目标营销

欺诈检测 – 银行、信用卡公司和在线商店使用图形分析来识别异常趋势。

供应链优化 – 公司可以使用图表分析来确定其供应链的最佳路线

贷款决策 – 抵押贷款公司和银行使用图表分析来评估申请人的数据以做出贷款决策。

Google 如何使用图形处理

Google 使用一种称为 PageRank 算法的图形分析算法。 PageRank 算法根据重要性对图中的顶点进行排名，其中重要性是指向该顶点的边数。该算法是由 Google 的创始人开发的，因此流行的搜索引擎是 PageRank 的一个典型例子。谷歌根据页面的重要性对页面进行排名，重要性是指向页面的超链接数量。

那么哪个更快？

也许最大的问题是——哪个更快？根据各种基准，这两个选项都显着提高了 MapReduce 性能；但是，获胜者可能取决于谁在进行测量。就独立第三方评估而言，陪审团仍未出局。

Spark 声称运行速度比 MapReduce 快 100 倍。在加州大学伯克利分校的 Amplab 进行的基准测试表明，它的运行速度比它的同类产品快得多（测试将 Spark 称为 Shark，它是 Spark SQL 的前身）。

然而，由于伯克利发明了 Spark，这些测试可能并非完全没有偏见。此外，这些基准测试是几年前使用运行在 MapReduce 上的 Hive 0.12 进行的。从版本 0.13 开始，Hive 使用 Tez 作为其执行引擎，从而显着提高了性能。

与此同时，Hortonworks 对两者之间的问题表现进行了基准测试。他们发现在 Tez 上运行的 Hive 0.13 的运行速度比 Hive 0.12 快 100 倍（尽管相当多的测试查询神秘地消失了）。快了 100 倍……嗯，听起来很熟悉？

因此，它们的性能都比 Hadoop MapReduce 高 100 倍。但哪个最快？

没有人可以说——或者更确切地说，他们不会承认。如果你问为 IBM 工作的人，他们会告诉你答案都不是，而且 IBM Big SQL 比两者都快。我们需要第三方来运行独立的性能测试并一劳永逸地确定分数。

小结：

这个问题最终可能归结为政治和受欢迎程度。这是大数据巨头的冲突，Cloudera 支持 Spark，Hortonworks 支持 Tez。 Spark 更为广泛，因为它可以在各种发行版中使用，而 Tez 仅在 Hortonworks 的发行版中可用。

最终，用户群可能决定框架的命运。目前，至少根据谷歌趋势，Spark 正在赢得这场比赛。

也许在炒作消退之后，在人们获得了更多与两者合作的经验之后，我们最终将能够判断谁将成为 MapReduce 王冠的继承人。

作者 east

分类归档大数据开发