east – 第54页 – gitweixin

Docker 3月 18,2022

解决Docker搭建kibana访问出现[Kibana server is not ready yet]的问题

在采用单机版的docker搭建elk，照着网上的教程如下：

1、下载es7.3.0镜像
docker pull docker.elastic.co/elasticsearch/elasticsearch:7.3.0

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

3、启动es的docker镜像
docker run --name es  -p 9200:9200 -p 9300:9300  --network esnet -e "discovery.type=single-node" bdaab402b220


4、下载docker安装es对应版本的kibana
docker run --name kibana --net esnet -e ELASTICSEARCH_URL=http://127.0.0.1:9200 -p 5601:5601 -d 8bcee4a4f79d

在这一步，满心欢喜想访问kibana后台。

http://localhost:5601

却发现提示
Kibana server is not ready yet

原因是因为 ELASTICSEARCH_URL 配置的应该是容器的 ip，而不是本地ip。

1、首先查看ElasticSearch的容器内部的ip

docker inspect es (es是 ElasticSearch 在容器内部的名称）

Docker容器中启动服务和直接在Linux环境下安装服务，会有ip上的区别，不在是服务器的物理ip, 而是容器对外暴露对的ip, 通过docker inspect elasticsearch查看ES容器暴露出来的ip

2、然后进入 Kibana 容器内部，修改 kibana.yml 中的ip

$ docker exec -it kibana容器id /bin/bash
$ cd config
$ vi kibana.yml
#
# ** THIS IS AN AUTO-GENERATED FILE **
#

# Default Kibana configuration for docker target
server.name: kibana
server.host: "0"
elasticsearch.hosts: [ "http://elasticsearch:9200" ]
xpack.monitoring.ui.container.elasticsearch.enabled: true

3需要将上面的 "http://elasticsearch:9200" 中的 elasticsearch 替换成上一步的es容器内部ip就可以了。

3、修改完成之后退出容器，重新启动即可

docker stop kibana
docker start kibana

重新访问http://localhost:5601,发现出来后台界面了。

作者 east

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

量化投资 2月 24,2022

波动产生利润，利用极限做T提高收益

我一个全职股民的前同事，最近3年股市平均收益率都超过20%，特别是2020年收益翻倍。

从他分享的炒股经验来看，除了他擅长选出好的股票，另一个盈利来源是他时不时把手中弱势股票卖出部分，把资金加到强势股，通过做T的方法降低了持仓成本，有时甚至提高收益几十个点。

以前一直痴迷做T，但不是做T失败变成被动加仓，就是T翻车看着大牛股翻倍。

通过不断总结教训，最近2年用极限做T的方法，不断取得了较好的正收益。

极限做T操作法则如下：

（1）做T时机和方向选择：就是早盘9：45分后，最好是10点后，通过初步分析出大盘和个股一天的走势。选择今天资金流入排名前，找出持仓中表现较强势的股票或ETF，特别是个股或ETF到了适合加仓时候，结合过去5天的平均振幅，分析出今天可能的最高价和最低价。如果个股当天表现弱势，大盘也不好，一般不要做正T，而是可以做反T。趋势向上的个股以做正T为主，趋势向下的以做反T为主。

（2）做T的仓位管理：在不是特别有把握时，用不高于原有仓位1/3的资金，最好把资金分成2等分，1等分买觉得能成交极限值（可以买比今天最低价高一点的价格），另一等分挂比前面更低，觉得成交性不大的。但股市的魔幻就是涨时涨过头，跌时跌过头。抓住这样的波动才能有更大的盈利。做反T的方法类似，在这里就不再重复。

（3）做T和做趋势结合：做T其实和抓住趋势不冲突。单边下跌趋势的，不适合做T。做T时保留好底仓，如果是ETF等好分割等分的，可以把每次做T加仓的1/5留下来。例如做T买1200，卖1000。加上选择趋势向上的股票或ETF，通过做T来从波动赚钱，同时盈利加仓，加大仓位来赚趋势上涨赚的钱。

（4）适合做T的标的：最好是结合前面文章说的ETF动量轮动来做。ETF不用印花税交易成本低，可以任意分割等分，极大提高做T成功率。ETF不好的地方是波动小，所以也可以结合一些小市值，高成长，估值合理估性活的个股。

作者 east

量化投资 2月 23,2022

改良版二八择时法则ETF轮动

这是在一本需要付费的量化投资策略上看的，觉得简单好用，是做量化投资好的入门。

首先说为什么选择ETF？

因为ETF基金在股票市场中生命力持久，甚至成为部分投资者的唯一选择，本质原因是它剥离了个股（及其背后的上市公司）特质风险，通过获得市场系统性风险反而降低了个股选择的难度，同时ETF基金和市场指数波动的吻合度在98%以上，追踪误差极小，这意味着我们基本上能够保证通过被动持有跑平市场。个股投资虽然有大幅度获利机会，但也非常容易被基准指数甩出很远跑输市场。ETF基金还有交易手续费低、交易规则多样、交易频率低等优势。

策略以上证50、沪深300、中证500、国债指数等相关4只ETF来轮动。基本思想是大盘指数好时，根据强者恒强的动力量因子，计算上证50、沪深300、中证500收盘价的20日涨幅，选取涨幅最大的；如果上面3个指数20日涨幅是负的，就买入国债指数来规避大盘风险。

上周我就是用类似策略，看到上证50、银行ETF走势强，其中银行ETF还是追高买的，但按照前面的网格交易的策略，都取到正的收益。

觉得上面的策略还是有点难取得超高收益，要收益高，还应该加入长期向上，近期资金流入的行业ETF。像上周买入的稀有金属ETF，走势秒杀我大部分持仓股。

对很多人来说，止损是比较难做到的，可以没在趋势走明显单边下跌，采用网格交易。把手中剩余资金重点分配给最强势的ETF。

作者 east

python, 量化投资 2月 20,2022

什么是ATR和在聚宽量化平台如何计算ATR

海龟们使用两种资金管理方法。首先，我们把头寸分成一个个小块。这样，即使一笔交易赔了钱，我们损失的也只是一个头寸的一部分。里奇和比尔把这些小块称作头寸单位。其次，我们使用里奇和比尔发明的一种创新性的头寸规模决定方法。这种方法以市场的每日上下波动为基础，而波动幅度是以不变美元价衡量的。他们会为每一个市场计算出一个特定的合约数量，目的是让所有市场的绝对波动幅度大致相等。里奇和比尔把他们的波动性指标称为N，尽管现在的人更习惯称它为真实波动幅度均值（average true range，ATR）。

在量化投资平台聚宽上面，可以用下面的代码来计算：

# 本文用于测算ATR，ATR是用于测算波动率的重要指标，其定义为：
# 1、昨日收盘价-当日最高价
# 2、昨日收盘价-当日最低价
# 3、当日最高价-当日最低价
# 三者的最大值
import pandas as pd

#显示所有列
pd.set_option('display.max_columns', None)
#显示所有行
pd.set_option('display.max_rows', None)
#设置value的显示长度为100，默认为50
pd.set_option('max_colwidth',100)

def get_ATR(stockname, begindate, enddate):
    # 获取行情数据
    df_stockdata = get_price(stockname, start_date=begindate, end_date=enddate, frequency='daily')
    df_stockdata = df_stockdata.reset_index()
    df_stockdata.rename(columns={'index':'stockdate'}, inplace=True)
#     print(df_stockdata)
    if type(df_stockdata) == int:
        print(stockname, '在这段时间内，没有数据！')
        return -1
    # 数据清理，首先要删掉空格行
    df_stockdata = df_stockdata.replace(0, np.nan)
    df_stockdata = df_stockdata.dropna()
    df_stockdata = df_stockdata.reset_index(drop=True)
    # 获取昨日收盘价
    df_yesterday_close = df_stockdata.loc[0:len(df_stockdata) - 2, 'close']
    df_yesterday_close.index = df_yesterday_close.index + 1
    df_yesterday_close.rename('yesterday_Close', inplace=True)  # 更新Series的名字，一定要加 inplace=True，否则改不过来
    # print(type(df_yesterday_close))
    df_stockdata = pd.concat([df_stockdata, df_yesterday_close], axis=1)
    df_stockdata = df_stockdata.dropna(axis=0)  # 删掉空值数据
    df_stockdata = df_stockdata.reset_index(drop=True)
    df_stockdata['ATR_3'] = (df_stockdata['high'] - df_stockdata['low'])  # 当日最高价-当日最低价
    df_stockdata['ATR_2'] = abs(df_stockdata['yesterday_Close'] - df_stockdata['low'])  # 昨日收盘价-当日最低价
    df_stockdata['ATR_1'] = abs(df_stockdata['yesterday_Close'] - df_stockdata['high'])  # 昨日收盘价-当日最低价
    # 重新构建一个临时DataFrame，用于取最大值
    df_temp_atr = df_stockdata[['ATR_3', 'ATR_2', 'ATR_1']]
    df_atr = df_temp_atr.max(axis=1)
    df_atr.rename('ATR', inplace=True)    # 重命名
    
    df_stockdata = pd.concat([df_stockdata, df_atr], axis=1)
    return df_stockdata

get_ATR('300144.XSHE', '2022-02-01', '2022-02-18')

作者 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