大数据开发 – 第23页

Hive 7月 20,2022

Hive批量删除表的工具

在建设数仓时，批量创建了很多表，发现创建错了，想批量删除，一个个删除觉得太麻烦，想批量删除。

可以利用hive打印出所有表名，用shell来搜索关键字，并用shell的管道技术来输出。

#/bin/sh
db_name=$1
#用于筛选出待删除表名的关键字
key_word=$2
hive -e "SHOW TABLES IN ${db_name};"|grep "${key_word}"|xargs -I{} echo drop table ${db_name}.{}\;>>batch_drop.log

然后打开batch_drop.log，就可以批量删除table的命令，复制到hive的shell命令界面中执行就可以。

作者 east

Hive 7月 14,2022

Hive增加字段casecade报“Error while compiling statement: FAILED: NullPointerException null”

最近在使用hive做数仓开发遇到各种坑，使用casecade为了刷新旧分区数据（
cascade的中文翻译为“级联”，也就是不仅变更新分区的表结构（metadata），同时也变更旧分区的表结构）

例如用下面的语句

ALTER TABLE plaintiff_info ADD COLUMNS(test3 STRING ) CASCADE

就遇到下面的错误：Error while compiling statement: FAILED: NullPointerException null

因为之前这样增加字段并刷新历史数据成功过，怀疑过字段为空等，各种怀疑和尝试，最后发现之前成功的是手动创建分区表，失败的是非分区表。非分区表再添加字段时不能加CASCADE 。

分区表，指创建时有

PARTITIONED BY (
dt string)

或者使用spark SQL创建带有 partitionBy(“year”)

作者 east

Spark 7月 12,2022

Idea java和scala混合打包遇到的各种坑

1、采用默认java的maven配置，运行时发现scala的包没打进去。

发现需要配置scala的编译器，网上有2种方式：

方式1：

 <plugin>
                <groupId>org.scala-tools</groupId>
                <artifactId>maven-scala-plugin</artifactId>
                <version>2.15.2</version>
                <configuration><</configuration>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                            <goal>testCompile</goal>
                        </goals>
                    </execution>
                </executions>

            </plugin>

方式2：

<plugin>
    <groupId>net.alchim31.maven</groupId>
    <artifactId>scala-maven-plugin</artifactId>
    <version>3.2.2</version>
    <configuration>
        <recompileMode>incremental</recompileMode>     
    </configuration>
    <executions>
        <execution>
            <goals>
                <goal>compile</goal>
                <goal>testCompile</goal>
            </goals>
        </execution>
    </executions>
</plugin>

2、编译时scala外部依赖包没找到

需要添加

<configuration>                
                    <args>
                        <!-- 编译时使用 libs 目录下的 jar 包，通过 mvn scala:help 查看说明 -->
                        <arg>-extdirs</arg>
                        <arg>D:/code/spark/jars</arg>
                    </args>
                </configuration>

完整配置如下：

 <plugin>
                <groupId>net.alchim31.maven</groupId>
                <artifactId>scala-maven-plugin</artifactId>
                <version>3.2.2</version>
                <configuration>
                    <recompileMode>incremental</recompileMode>
                    <args>
                        <!-- 编译时使用 libs 目录下的 jar 包，通过 mvn scala:help 查看说明 -->
                        <arg>-extdirs</arg>
                        <arg>D:/code/spark/jars</arg>
                    </args>
                </configuration>
                <executions>
                    <execution>
                        <goals>
                            <goal>compile</goal>
                            <goal>testCompile</goal>
                        </goals>
                    </execution>
                </executions>
            </plugin>

3、maven打包时提示内存不足

在如图位置,VM Options增加内存就可以

作者 east

Hive, 数据挖掘 7月 1,2022

Datart连接Hive需要的jar包

最近在做数仓开发，把结果输入到datart这个bi工具，刚开始用datart来连接mysql，为了减少层次，想直接连接hive。发现是需要配置jdbc驱动的。

大数据集群用的是CDH6.3.2，需要的jar包跟网上不同。刚开始用hive-jdbc-2.1.1-cdh6.3.2-standalone.jar，发现冲突了。后来从用下面这些包解决了依赖问题。

不过用最新的 datart-1.0.0-beta.3 ，发现存在问题：
无法查询sql 。报错如下：

Request failed with status code 400

github官方论坛提示：该问题已经修复，等beta.4版本发布后升级即可

作者 east

Hadoop, Hive 6月 24,2022

idea远程连接hadoop、hive操作权限不够问题解决

使用idea远程操作hive，进行写数据时提示权限不够：

Permission denied: user=test, access=WRITE, inode=”/data/warehouse/db1/dws/dws_test1″:hdfs:supergroup:drwxr-xr-x

要解决这类问题，最方便方法是hdfs设置HADOOP_USER_NAME，这样远程访问就用指定的名来访问。（如果是CDH安装的，超级用户hdfs ）有几类操作方法：

方法一
打开要设置类文件的Configurations

第二步name中写自己的用户名称就行

再重新运行java文件就可以了

方法二
在java文件中进行设置
FileSystem fs= FileSystem.get(new URI(“hdfs://cdh01:9000”),conf,“hdfs”);

方法三

设置windows环境变量

作者 east

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

发现原因是在于，建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

分类归档大数据开发