大数据开发 – 第26页

Docker 3月 18,2022

Docker网络名词解释和例子

最近在学习docker，研究如何用docker安装elk。看到有一个教程是这样说的：

docker创建一个网络，方便elk使用
docker network create esnet

那么网络是什么？

Docker在容器内部运行应用，这些应用之间的交互依赖于大量不同的网络。Docker对于容器之间、容器与外部网络和VLAN之间的连接均有相应的解决方案。在顶层设计中，Docker网络架构由3个主要部分构成：CNM、Libnetwork和驱动。

Docker网络架构的设计规范是CNM。网络就是需要交互的终端的集合，并且终端之间相互独立。Docker环境中最小的调度单位就是容器，沙盒被放置在容器内部，为容器提供网络连接。

容器A只有一个接口（终端）并连接到了网络A。容器B有两个接口（终端）并且分别接入了网络A和网络B。容器A与B之间是可以相互通信的，因为都接入了网络A。但是，如果没有三层路由器的支持，容器B的两个终端之间是不能进行通信的。

每个Docker主机都有一个默认的单机桥接网络。在Linux上网络名称为bridge，在Windows上叫作nat。除非读者通过命令行创建容器时指定参数–network，否则默认情况下，新创建的容器都会连接到该网络。

查看docker网络的命令：

docker network ls

作者 east

spring 3月 17,2022

Spark大数据平台调度任务的优化

在大数据平台生产环境上，遇到一个头疼的问题，每天都要定时运行一个任务。刚开始数据量小和简单，用cron来定时调用可以满足需求。

后来数据量大，出现昨天的任务没跑完，今天的任务又要开始了，在大数据平台上运行的任务越来越多，大数据平台资源被占满了。

考虑进行下面的优化：

1、某个任务如果运行超过30小时时，进行中断。

2、采用DelayQueue来实现延时队列，等前面的任务执行完或被中断，又到它的开始时间时才进行执行。

延时队列相比于普通队列最大的区别就体现在其延时的属性上，普通队列的元素是先进先出，按入队顺序进行处理，而延时队列中的元素在入队时会指定一个延迟时间，表示其希望能够在经过该指定时间后处理。

入门例子

DelayQueue 非常适合指定时间之后，才能让消费者获取到的场景。

private static class DelayElem implements Delayed {
    /**
     * 延迟时间
     */
    private final long delay;
    /**
     * 到期时间
     */
    private final long expire;
    /**
     * 数据
     */
    private final String msg;

    private DelayElem(long delay, String msg) {
        this.delay = delay;
        this.msg = msg;
        //到期时间 = 当前时间+延迟时间
        this.expire = System.currentTimeMillis() + this.delay;
    }
    /**
     * 需要实现的接口，获得延迟时间
     *
     * 用过期时间-当前时间
     * @param unit 时间单位
     * @return 延迟时间
     */
    @Override
    public long getDelay(TimeUnit unit) {
        return unit.convert(this.expire - System.currentTimeMillis() , TimeUnit.MILLISECONDS);
    }
    /**
     * 用于延迟队列内部比较排序
     * <p>
     * 当前时间的延迟时间 - 比较对象的延迟时间
     *
     * @param o 比较对象
     * @return 结果
     */
    @Override
    public int compareTo(Delayed o) {
        return (int) (this.getDelay(TimeUnit.MILLISECONDS) - o.getDelay(TimeUnit.MILLISECONDS));
    }
    @Override
    public String toString() {
        return "DelayElem{" +
                "delay=" + delay +
                ", expire=" + expire +
                ", msg='" + msg + '\'' +
                '}';
    }
}

private static class WriteThread extends Thread {
    private final DelayQueue<DelayElem> delayQueue;
    private WriteThread(DelayQueue<DelayElem> delayQueue) {
        this.delayQueue = delayQueue;
    }
    @Override
    public void run() {
        for(int i = 0; i < 3; i++) {
            try {
                TimeUnit.MILLISECONDS.sleep(100);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
            DelayElem element = new DelayElem(1000,i+"test");
            delayQueue.offer(element);
            System.out.println(System.currentTimeMillis() + " 放入元素 " + i);
        }
    }
}

private static class ReadThread extends Thread {
    private final DelayQueue<DelayElem> delayQueue;
    private ReadThread(DelayQueue<DelayElem> delayQueue) {
        this.delayQueue = delayQueue;
    }
    @Override
    public void run() {
        while (true){
            try {
                DelayElem element =  delayQueue.take();
                System.out.println(System.currentTimeMillis() +" 获取元素：" + element);
            } catch (InterruptedException e) {
                e.printStackTrace();
            }
        }
    }
}

public static void main(String[] args) throws InterruptedException {
    DelayQueue<DelayElem> delayQueue = new DelayQueue<>();
    new WriteThread(delayQueue).start();
    new ReadThread(delayQueue).start();
}

作者 east

运维 3月 10,2022

如何恢复MySQL误删除数据文件

在生产环境上，误删除ibdata1、ib_logfile0和ib_logfile1。如果mysql进程还在运行，此时select读取数据还是insert、update更新数据都正常。

这时可以通过命令查看mysql的进程

# ps -ef | grep mysqld|awk '{print $1,$2}'|grep mysql
# mysql 4070

查看mysql误删除文件在proc的位置

ll /proc/4070fd|grep -e ibdata -e ib_

可以看到，被删除的3个文件ibdata1、ib_logfile0和ib_logfile1在内存中已经标记为deleted了，可见文件确实是被删除了。

那么为什么MySQL还能正常使用呢？其实，mysqld在运行状态下，会保持ibdata1、ib_logfile0、ib_logfile1这些文件为打开状态，即使把它们删除了，它们仍旧存在于内存文件系统中，所以，mysqld仍然可以对其进行读写。只要mysqld进程不结束（MySQL服务不重启），就可以通过proc文件系统找到这几个被删除的文件。

现在数据库还一直对外提供服务，也就是有数据会持续写入，而在InnoDB存储引擎的buffer pool中，有许多dirty page（脏数据，就是内存中的数据已经被修改，但是没有写到磁盘中）还没提交，如果直接把文件复制回去，肯定会丢失数据，甚至还有可能导致ibdata1文件损坏。在复制数据文件之前，必须保证所有buffer pool中的数据修改都保存到了硬盘上，因此，首先需要停止目前的写入、更新和删除等操作，然后刷新内存数据到磁盘，最后才能复制文件。如何操作呢？可执行下面几个SQL：

FLUSH TABLES WITH READ LOCK;
SHOW engine innodb STATUS\G;
show variables like '%innodb_max_dirty_pages_pct%';
SET global innodb_max_dirty_pages_pct=0;
这样设置后，脏页会迅速减少，磁盘写操作会迅速完成。等待所有脏数据刷新到磁盘后，就可以进行文件复制了。

cp /proc/4070/fd/10  /data1/mysql/ib_logfile1
cp /proc/4070/fd/4   /data1/mysql/ibdata1
cp /proc/4070/fd/9   /data1/myql/ib_logfile0

修改文件权限为MySQL，操作如下：

chown mysql:mysql /data1/mysql/ib*

所有操作完成后，还需要重启数据库服务：

/etc/init.d/mysqld restart

作者 east

Java, python 3月 6,2022

Python基础语法规则和Java不同的地方

Java是现在最流行的语言，也是广大程序员最熟悉的语言，而Python作为在人工智能领域的新星，通过对比Java语言来学习Python语言，可以起到事半功倍的效果。

和Java单行注释不同，Python注释更像shell等脚本语言， python语言单行注释通常是以“#”号开头，在“#”号后面紧跟注释说明的文字。

Java语言用{ } 来区分代码块， Python是用缩进代码，缩进相同的一组语句构成一个代码块，也称为代码组。

在数学运算上，Java使用除法，如果除数和被除数都是整数，那么结果还是整数。如果结果要为准确的浮点数，要对其中一个数进行强制转换。Python就没有那么麻烦。 Python数值的除法包含两个运算符：“/”返回一个浮点数，“//”返回一个整数，在混合计算时，Python会把整数转换成为浮点数。

在字符串的定义上，Python更加灵活。可以使用单引号（’）、双引号（”）或三引号（”’或”””）来标识字符串，引号的开始与结束必须是相同类型的。

Python中没有switch和case语句，多路分支语句只能通过if…elif…else流程控制语句来实现，并且每个分支语句都要有严格的缩进。 for循环中也可以使用else语句。

作者 east

Spark 2月 25,2022

Spark Streaming调用http接口导致卡住了

在生产环境上，之前一直正常，最近运行Spark Streaming出不来结果，通过打印日志和远程调试，发现是调用一个接口，是用HttpUrlConnection来请求，在下面的语句一直没返回结果导致卡住了。

connection.getResonseCode();

解决方法，是对
HttpUrlConnection 设置超时时间

HttpURLConnection connection= (HttpURLConnection)url.openConnection();
connection.setConnectTimeout(30000);
connection.setReadTimeout(30000);

作者 east

flume, Spark 2月 20,2022

大数据运维一些常见批量操作命令

在使用flume过程，由于故障停止采集，堆积文件很多，想迁移到新目录，但如果文件数目太多的话，想直接用mv 命令会报错。这时我们需要利用管道技术和xargs命令。

xargs（英文全拼： eXtended ARGuments）是给命令传递参数的一个过滤器，也是组合多个命令的一个工具。

xargs 可以将管道或标准输入（stdin）数据转换成命令行参数，也能够从文件的输出中读取数据。

常用操作1：把当前目录的文件移到新的目录newdir

find . -name '2106*.json' | xargs -i mv {} ../newdir

常用操作2：把当前目录json.1后缀的批量删除

find . -name '*.json.1' | xargs -i rm -f {}

常用操作3：shell批量kill掉java进程

ps aux | grep test.jar | grep -v grep | awk '{print $2}' | xargs kill -9

作者 east

Hive, 数据仓库 2月 19,2022

Hive构建数据仓库常用的函数

concat（）函数。

concat（）函数用于连接字符串，在连接字符串时，只要其中一个字符串是NULL，结果就返回NULL。

concat_ws（）函数。

concat_ws（）函数同样用于连接字符串，在连接字符串时，只要有一个字符串不是NULL，结果就不会返回NULL。concat_ws（）函数需要指定分隔符。

str_to_map（）函数。

● 语法描述。str_to_map(VARCHAR text,VARCHAR listDelimiter,VARCHARkeyValueDelimiter)。

● 功能描述。使用listDelimiter将text分隔成key-value对，然后使用keyValueDelimiter分隔每个keyvalue对，并组装成MAP返回。默认listDelimiter为“，”，keyValueDelimiter为“=”。

nvl（）函数

基本语法：nvl（表达式1，表达式2）。如果表达式1为空值，则nvl（）函数返回表达式2的值，否则返回表达式1的值。nvl（）函数的作用是把一个空值（null）转换成一个实际的值。其表达式的数据类型可以是数字型、字符型和日期型。需要注意的是，表达式1和表达式2的数据类型必须相同。

日期处理函数

1）date_format（）函数（根据格式整理日期）

hive> select date_format('2020-03-18',''yyyy-MM');
hive> 2020-03

2）date_add（）函数（加减日期）

hive> select date_add('2020-03-11',1);
hive> 2020-03-12

3）next_day（）函数

（1）获取当前日期的下一个星期一。

hive> select next_day('2020-03-13','MO');
hive> 2020-03-16

（2）获取当前周的星期一。

hive> select date_add(next_day('2020-03-13','MO'),-7);
hive> 2020-03-11

4）last_day（）函数（获取当月最后一天的日期）

作者 east

Hive 2月 19,2022

Tez运行报TezSeeion has already shutdown错误

运行Tez出现下面错误：

Caused by: org.apache.tez.dag.api.SessionNotRunning: TezSeeion has already shutdown.Application application_15xxxx  failed 2 times due to AM Container for appattempt_xxx exited with exitCode: -103 For more detailed output,check application tracking page:http://hadoop1:8088/cluster/app/application_15xxxThen,click on links to logs of each attempt.

产生原因：

这是由于Container使用过多内存而被NodeManager杀死进程

解决方法：

修改Hadoop的配置文件yarn-site.xml，增加如下配置，关掉虚拟内存检查，修改后，分发配置文件，并重启集群。

<property>
   <name>yarn.nodemanager.vmem-check-enabled</name>
   <value>false</value>
</property>

作者 east

Spark 1月 6,2022

Spark如何在生产环境调试

接手别人开发的spark程序，大概弄懂整个流程，但一些细节总是猜不透，在生产环境运行效果也达不到理想。

想去修改，遇到下面的问题：

一、由于生产环境是运行在linux服务上的，在华为HD Insight大数据平台上，在开发机不知怎样调试。

解决方式：后来发现其实在idea是可以远程调试的：

打开工程，在菜单栏中选择“Run > Edit Configurations”。
在弹出的配置窗口中用鼠标左键单击左上角的号，在下拉菜单中选择Remote，如图1所示。图1 选择Remote

3. 选择对应要调试的源码模块路径，并配置远端调试参数Host和Port，如图2所示。
其中Host为Spark运行机器IP地址，Port为调试的端口号（确保该端口在运行机器上没被占用）。

说明：当改变Port端口号时，For JDK1.4.x对应的调试命令也跟着改变，比如Port设置为5006，对应调试命令会变更为-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=5006，这个调试命令在启动Spark程序时要用到。

4.执行以下命令，远端启动Spark运行SparkPi。 ./spark-submit –master yarn-client –driver-java-options “-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=5006” –class org.apache.spark.examples.SparkPi /opt/FI-Client/Spark2x/spark/examples/jars/spark-examples_2.11-2.1.0.jar 用户调试时需要把–class和jar包换成自己的程序，-Xdebug -Xrunjdwp:transport=dt_socket,server=y,suspend=y,address=5006需要换成3获取到的For JDK1.4.x对应的调试命令。

5.设置调试断点。在IDEA代码编辑窗口左侧空白处单击鼠标左键设置相应代码行断点，如图4所示，在SparkPi.scala的29行设置断点。

6.启动调试。在IDEA菜单栏中选择“Run > Debug ‘Unnamed’”开启调试窗口，接着开始SparkPi的调试，比如单步调试、查看调用栈、跟踪变量值等，如图5所示。

二、在spark executor执行的，如果看调试结果？

解决方式是在相应的rdd加上collect()方法，把结果传送到driver来看

作者 east

Java 12月 14,2021

Java多线程使用总结

Java线程的3种创建方式，使用继承方式的好处是方便传参，你可以在子类里面添加成员变量，通过set方法设置参数或者通过构造函数进行传递，而如果使用Runnable方式，则只能使用主线程里面被声明为final的变量。不好的地方是Java不支持多继承，如果继承了Thread类，那么子类不能再继承其他类，而Runable则没有这个限制。前两种方式都没办法拿到任务的返回结果，但是Futuretask方式可以。

sleep与yield方法的区别在于，当线程调用sleep方法时调用线程会被阻塞挂起指定的时间，在这期间线程调度器不会去调度该线程。而调用yield方法时，线程只是让出自己剩余的时间片，并没有被阻塞挂起，而是处于就绪状态，线程调度器下一次调度时就有可能调度到当前线程执行。

守护线程并非只有虚拟机内部可以提供，用户也可以手动将一个用户线程设定/转换为守护线程。在Thread类中提供了一个setDaemon(true)方法来将一个普通的线程（用户线程）设置为守护线

public final void setDaemon(boolean on);

如果你希望在主线程结束后JVM进程马上结束，那么在创建线程时可以将其设置为守护线程，如果你希望在主线程结束后子线程继续工作，等子线程结束后再让JVM进程结束，那么就将子线程设置为用户线程。

作者 east

分类归档大数据开发