大数据开发 – 第26页

Elasticsearch, neo4j, solr 3月 22,2022

neo4j、solr、es数据同步和增量更新

一、Neo4j的数据增量更新：

图数据除了节点的增量更新还牵扯到边的增量更新，节点其实还好说，无论是新增节点还是新增节点属性或者是节点属性更新，实际上都是在节点表可以完成，不是很消耗资源；比较复杂且影响更新效率的其实边的增量更新。

1、节点更新

这个需求其实很普遍，比如我有一个节点：

(n:Person {id: 'argan', name: 'argan', age: 32})

然后用户又传递了一个人数据过来：

{id: 'argan', age: 30, sex: 'male', email: 'arganzheng@gmail.com'}

可以看到更新了一个属性：年龄，新增了两个属性：性别和电子邮件。我们希望最后的结果是：

(n:Person {id: 'argan', name: 'argan', age: 30, sex: 'male', email: 'arganzheng@gmail.com'})。

需要注意的是name是没有传递的，但还是保留着的。如果要删除一个属性，需要把它的值显式的设置为空。

在Neo4j的要怎么做到呢？

Neo4j的提供了合并语句来实现这个功能。

与ON CREATE和ON MATCH合并

如果需要创建节点，请合并节点并设置属性。

MERGE (n:Person { id: 'argan' })
ON CREATE SET n.created = timestamp()
ON MATCH SET n.lastAccessed = timestamp()
RETURN n.name, n.created, n.lastAccessed

上面的例子可以这么写：

MERGE (n:Node {id: 'argan'}) SET n += {id: 'argan', age: 30, sex: 'male', email: 'arganzheng@gmail.com'} 
RETURN n

因为这里采用了+=本身就是合并属性，所以区分不需要的英文ON CREATE还是ON MATCH。

同样关系也可以用合并保证只创建一次：

MATCH (n), (m) WHERE n.id = "argan" AND m.id = "magi" CREATE (n)-[:KNOWS]->(m)

写成这样子就可以保证唯一了：

MATCH (n:User {name: "argan"}), (m:User {name: "magi"}) MERGE (n)-[:KNOWS]->(m)

2、neo4j如何支持动态节点标签和关系类型？

上面的合并语句能够实现“存在更新，否则创建”的逻辑，但是还有一个问题没有解决，就是没有设置节点的标签。我们希望构建的节点数据完全是运行时根据用户提供的数据构造的，包括。标签比如用户提供如下数据：

:param batch: [{properties: {name: "argan", label: "Person", id: "1", age: 31}}, {properties: {name: "magi", label: "Person", id: "2", age: 28}}]

下面的暗号语句并没有设置节点的标签，虽然节点有一个叫做标签的属性：

UNWIND {batch} as row 
MERGE (n {id: row.id})
SET n += row.properties

那么我们能不能简单的指定标签呢？

UNWIND {batch} as row 
MERGE (n:row.properties.label {id: row.id})
SET n += row.properties

但是遗憾的是这个语句会报错，因为neo4j不支持动态的节点标签。把row.properties.label去掉或者改成一个固定的字符串就没有问题。

改成这样子也不行：

UNWIND {batch} as row   MERGE (n {id: row.id} )   SET n:row.properties.label,  n += row.properties

绑定变量也不行：

UNWIND {batch} as row   MERGE (n {id: row.id} )   SET n:{label},  n += row.properties

直接指定标签就可以了：

UNWIND {batch} as row   MERGE (n {id: row.id} )   SET n:Test,  n += row.properties

也就是说3.3.13.9。在节点上设置标签也并不支持动态标签..

笔记

neo4j的设置标签还有一个问题，就是它其实是新增标签，不是修改标签。要到更新的效果，你需要先删除掉，再新增..

MATCH (n) WHERE ID(n) = 14  REMOVE n:oldLabel SET n:newLabel

如果是单条数据更新，那其实很简单，我们只需要做字符串拼接就可以了：

String label = vertex.getLabel(); "MERGE (n:" + label + " {id: {id}} " + "SET n += {properties}"

但是关键是我们这里是在Neo4j的内部用开卷展开的服务端变量，如果它不允许动态变量，根本搞不定。难道真的要一条条的插入，那会非常慢的！Neo4j的的插入性能是众所周知的差。一种做法就是先批量插入数据，设置一个临时的标签，然后再批量的更新标签。不过需要两次操作，性能肯定至少慢两倍。

有没有什么方式呢？谷歌了很久，发现了也有人遇到这样的问题：功能请求：apoc支持MERGE节点和rels＃271和是否可以使用数据驱动的节点或关系标签进行合并？。

原理跟单条数据插入一样，只是由于退绕是在服务端（Neo4j的）进行的，所以拼接也只能在服务端进行，怎么拼接的就是用？apoc.cypher.doIt拼接后让它在服务端执行：

UNWIND {batch} as row  WITH 'MERGE (n:' + row.properties.label + ' { id: row.id }) SET n += row.properties return n' AS cypher CALL apoc.cypher.doIt(cypher, {}) YIELD value return value.n

但是可惜，会报这样的异常：

org.neo4j.driver.v1.exceptions.ClientException: 
Failed to invoke procedure `apoc.cypher.doIt`: 
Caused by: org.neo4j.graphdb.QueryExecutionException: 
Variable `row` not defined (line 1, column 23 (offset: 22)) "MERGE (n:Person { id: row.id }) SET n += row.properties return n"

所以还是要分两步进行，不过可以合并在一起SET标签：传递标签名称作为参数：

UNWIND {batch} as row  MERGE (n { id: row.id }) 
SET n += row.properties  WITH n  CALL apoc.create.addLabels(id(n), [n.label]) 
YIELD node RETURN node

这样就可以正确的保存数据并且动态设置标签了。笔

本来我们是可以直接使用APOC库的apoc.merge.node状语从句：apoc.create.relationship动态的更新节点标签，关系和节点的。但是正如前面分析的，apoc.merge.node状语从句：apoc.create.relationship现在的实现其实的英文一个防重复CREATE而已，不能达到更新的目的。否则我们的实现将非常简单明了：

更新节点：

UWNIND {batch} as row CALL apoc.merge.node(row.labels, {id: row.id} , row.properties) 
yield node RETURN count(*)

更新关系：

UWNIND {batch} as row MATCH (from) WHERE id(from) = row.from 
MATCH (to:Label) where to.key = row.to CALL apoc.merge.relationship(from, row.type, {id: row.id}, row.properties, to) 
yield rel RETURN count(*)

一种做法就是叉一个分支出来，修改源码，部署自己的罐子包。

二、solr 的增量更新

1.首先要弄懂几个必要的属性，以及数据库建表事项，和dataimporter.properties 、data-config.xml里面的数据

<!– transformer 格式转化：HTMLStripTransformer 索引中忽略HTML标签   —>
<!– query:查询数据库表符合记录数据   —>
<!– deltaQuery:增量索引查询主键ID    —>    注意这个只能返回ID字段
<!– deltaImportQuery:增量索引查询导入数据 —>
<!– deletedPkQuery:增量索引删除主键ID查询 —> 注意这个只能返回ID字段

2.数据库配置注意事项

1.如果只涉及添加，与修改业务，那么数据库里只需额外有一个timpstamp字段
就可以了，默认值为当前系统时间，CURRENT_TIMESTAMP
2.如果还涉及删除业务，那么数据里就需额外再多添加一个字段isdelete，int类型的
用0,1来标识，此条记录是否被删除

3.dataimporter.properties

这个配置文件很重要，它是用来记录当前时间与上一次修改时间的，通过它能够找出，那些，新添加的，修改的，或删除的记录标识，此条记录是否被删除的记录

4.增量更新就是在全量更新的基础上加上一些配置，配置如下：

<?xml version="1.0" encoding="UTF-8" ?>
<dataConfig> 
    <!--数据源-->
    <dataSource type="JdbcDataSource"
                driver="com.mysql.jdbc.Driver"
                url="jdbc:mysql://192.168.2.10:3306/xtjkqyfw"
                user="root"
                password="Biaopu8888"/>
    <document> 

        <entity name="solrTest" 
        query="SELECT fid,ftime,fcontent,ftitle,flastupdatetime FROM solrTest where flag = '0'"
        deltaImportQuery = "SELECT fid,ftime,fcontent,ftitle,flastupdatetime FROM solrTest where fid = '${dataimporter.delta.fid}'"
        deltaQuery = "SELECT fid FROM solrTest where flastupdatetime > '${dataimporter.last_index_time}' and flag = '0'"
        deletedPkQuery = "SELECT fid FROM solrTest where flag = '1'"
        >
            <!--查询的数据和数据库索引意义对应column 是查询的字段name 是solr索引对应的字段-->
            <field column="fid" name="fid"/>
            <field column="ftitle" name="ftitle"/>
            <field column="fcontent" name="fcontent"/>
            <field column="flastupdatetime" name="flastupdatetime"/>
            <field column="ftime" name="ftime"/>
        </entity>
        
    </document> 
</dataConfig>

三、LOGSTASH-INPUT-JDBC 实现数据库同步ES

在数据方面碰到第一个问题是怎么将postgres中的数据库中同步到es中，在网上找了下相关文档，只有logstash-input-jdbc这个插件还在维护，而且在es中logstash高版本已经集成了这一插件，所以就省去了安装ruby和安装插件这一步了

1 安装elasticsearch logstash kibana 三件套

2 下载数据库驱动

图方便的话可以直接拷贝maven仓库里面的即可

3 添加 .conf文件

input {  
  jdbc {     
 # mysql 数据库链接,shop为数据库名     
 jdbc_connection_string => "jdbc:postgresql://ip:5432/chongqing_gis"     
 # 用户名和密码     
 jdbc_user => ""    
  jdbc_password => ""    
  # 驱动     
 jdbc_driver_library => "E:/ES/logstash-7.8.0/postgres/postgresql-42.2.9.jar"     
 # 驱动类名     
 jdbc_driver_class => "org.postgresql.Driver"      jdbc_paging_enabled => "true"   
   jdbc_page_size => "50000"     
 # 执行的sql 文件路径+名称      
statement_filepath => "E:/ES/logstash-7.8.0/postgres/jdbc.sql"      # 设置监听间隔  各字段含义（由左至右）分、时、天、月、年，全部为*默认含义为每分钟都更新     
 # schedule => "* * * * *"  
  }
}
 filter {  
  json {    
    source => "message"  
      remove_field => ["message"]   
 }
} 
output {   
 elasticsearch {   
     # ES的IP地址及端口      
  hosts => ["localhost:9200"]     
   # 索引名称      
  index => "test_index"    
    # 需要关联的数据库中有有一个id字段，对应类型中的id        document_id => "%{gid}"   
 }  
  stdout {  
      # JSON格式输出    
    codec => json_lines 
   }
}

修改输入参数：数据库ip、端口、账号、密码等

修改输出参数：es的ip、端口、索引、document_id等

输出参数中索引如果还没有创建，在启动logstash时会自动根据默认模板创建索引,其中有些教程中出现了index_type的设置，这个类型在es的高版本中已经取消，不需要再设置，如果设置了，启动logstash会报错

statement_filepath 保存准备执行的sql文件

jdbc.sql文件：

select gid,name,address from qtpoi

在qtpoi表中有个字段是保存的空间地理信息geom字段，当我加上这个时，启动logstash一直报错，可能对空间字段需要做进一步的处理

Exception when executing JDBC query {:exception=>#<Sequel::DatabaseError: Java::OrgLogstash::MissingConverterException: Missing Converter handling for full class name=org.postgresql.util.PGobject, simple name=PGobject

4 启动logstash即可同步

bin/logstash -f config/postgres.conf

5 打开kibana即可查看到刚同步的数据

GET test_index/_doc/_search

6 如果设置了定时任务，logstash会定时去访问数据同步到es中，但是上面jdbc.sql文件中获取的是整张表的数据，也就是说每次同步都会对全表进行同步，但是我的需求是只需要第一次同步整张表后面只对更新的和修改的数据做同步，

在网上找了下，思路大概是给表增加一个新的字段，保存当前创建的时间或者是当前更新的时间，然后根据:sql_last_value这个函数获取大于这个函数的时间，就过滤出新增的数据和更新的数据，实现对增量数据同步到es，:sql_last_value这个函数官网也没说的太清楚，我大致认为是最后一个更新的值或者最后一次更新的时间

作者 east

flume 3月 19,2022

Logstash和flume全方位对比

Logstash架构如下：

Flume架构如下：

首先从结构对比，我们会惊人的发现，两者是多么的相似！Logstash的Shipper、Broker、Indexer分别和Flume的Source、Channel、Sink各自对应！只不过是Logstash集成了，Broker可以不需要，而Flume需要单独配置，且缺一不可，但这再一次说明了计算机的设计思想都是通用的！只是实现方式会不同而已。

从程序员的角度来说，上文也提到过了，Flume是真的很繁琐，你需要分别作source、channel、sink的手工配置，而且涉及到复杂的数据采集环境，你可能还要做多个配置，这在上面提过了，反过来说Logstash的配置就非常简洁清晰，三个部分的属性都定义好了，程序员自己去选择就行，就算没有，也可以自行开发插件，非常方便。当然了，Flume的插件也很多，但Channel就只有内存和文件这两种（其实现在不止了，但常用的也就两种）。读者可以看得出来，两者其实配置都是非常灵活的，只不过看场景取舍罢了。

其实从作者和历史背景来看，两者最初的设计目的就不太一样。Flume本身最初设计的目的是为了把数据传入HDFS中（并不是为了采集日志而设计，这和Logstash有根本的区别），所以理所应当侧重于数据的传输，程序员要非常清楚整个数据的路由，并且比Logstash还多了一个可靠性策略，上文中的channel就是用于持久化目的，数据除非确认传输到下一位置了，否则不会删除，这一步是通过事务来控制的，这样的设计使得可靠性非常好。相反，Logstash则明显侧重对数据的预处理，因为日志的字段需要大量的预处理，为解析做铺垫。

为什么先讲Logstash然后讲Flume？这里面有几个考虑，

其一：Logstash其实更有点像通用的模型，所以对新人来说理解起来更简单，而Flume这样轻量级的线程，可能有一定的计算机编程基础理解起来更好；

其二：目前大部分的情况下，Logstash用的更加多，这个数据我自己没有统计过，但是根据经验判断，Logstash可以和ELK其他组件配合使用，开发、应用都会简单很多，技术成熟，使用场景广泛。相反Flume组件就需要和其他很多工具配合使用，场景的针对性会比较强，更不用提Flume的配置过于繁琐复杂了。

作者 east

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

Springboot接口使用注意事项

Spring boot接口传参，如果接口用到整型，最好用Integer类型而不是int类型，例如

public ReturnResult deleteByPrimaryKey(@RequestParam("id") Integer id)

如果接口传参用int类型，最好带有默认值，例如：

@RequestParam(value="pageNum", required=false, defaultValue="1") int pageNum,
                                                @RequestParam(value="pageSize", required=false, defaultValue="10") int pageSize

给传参的接口，最好统一封装带有状态码和状态信息，这样客户端调用接口时，方便知道异常信息。


public class ReturnResult {

    //1是成功，0是失败
	private int code;
	
	private String msg;
	
	private Object data;
	
	
	
	

	public ReturnResult() {
		super();
	}


	public ReturnResult(int code, String msg, Object data) {
		super();
		this.code = code;
		this.msg = msg;
		this.data = data;
	}
	

	public ReturnResult(int code, String msg) {
		super();
		this.code = code;
		this.msg = msg;
	}

	public int getCode() {
		return code;
	}

	public void setCode(int code) {
		this.code = code;
	}

	public String getMsg() {
		return msg;
	}

	public void setMsg(String msg) {
		this.msg = msg;
	}

	public Object getData() {
		return data;
	}

	public void setData(Object data) {
		this.data = data;
	}
	
	
	
}

作者 east

solr 3月 17,2022

封装HttpSolrServer设置超时时间

凡是要连接网络的，都要设置超时时间，这样防止网络卡住了没返回结果，例子可以参考 Spark Streaming调用http接口导致卡住了

/**
 * HttpSolrServer扩展对象服务 .<br>
 * 专门设置请求连接超时时间和socket超时时间，调用solr请求服务，防止网络断开时无法中断调用线程<br>
 * 
 */
public class HttpSolrExServer extends HttpSolrServer {

    /** serialVersionUID */
    private static final long serialVersionUID = 4068028650816903817L;

    /**
     * 连接超时值
     */
    private static Integer connectionTimeoutNum;

    /**
     * socket超时值
     */
    private static Integer socketTimeoutNum;



    /**
     * 有参构造函数
     * 
     * @param baseUrl url地址
     */
    public HttpSolrExServer(String baseUrl) {
        super(baseUrl);
        // 设置请求连接超时时间，以毫秒为单位
        this.setConnectionTimeout(getConnectionTimeout());
        // 设置socket超时时间，以毫秒为单位
        this.setSoTimeout(getSocketTimeout());
    }

    public HttpSolrExServer(String baseUrl, HttpClient client){
        super(baseUrl,client);
        // 设置请求连接超时时间，以毫秒为单位
        this.setConnectionTimeout(getConnectionTimeout());
        // 设置socket超时时间，以毫秒为单位
        this.setSoTimeout(getSocketTimeout());
    }

    /**
     * 得到solr的连接超时时间，以毫秒为单位，默认为60秒<br>
     * 
     * @return
     */
    private static int getConnectionTimeout() {
        if (connectionTimeoutNum == null || connectionTimeoutNum <= 0) {
            // solr的连接超时时间，以毫秒为单位，默认为60秒
            int defaultConnectionTimeout = 60000;
            // solr的连接超时时间字符串变量值

            String connectionTimeoutStr = ConfigUtil.getPropsValueByKey("ac.httpSolr.connectionTimeout");
            

            try {
                int configValue = Integer.parseInt(connectionTimeoutStr);

                if (configValue > 0) {
                    defaultConnectionTimeout = configValue;
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
            connectionTimeoutNum = defaultConnectionTimeout;
        }
        return connectionTimeoutNum.intValue();
    }

    /**
     * 得到solr的socket超时时间，以毫秒为单位，默认为60秒<br>
     * 
     * @return
     */
    private static int getSocketTimeout() {
        if (socketTimeoutNum == null || socketTimeoutNum <= 0) {
            // solr的socket超时时间，以毫秒为单位，默认为60秒
            int defaultSocketTimeout = 60000;
            // solr的socket超时时间字符串变量值
            String socketTimeoutStr = ConfigUtil.getPropsValueByKey("ac.httpSolr.socketTimeout");

            try {
                int configValue = Integer.parseInt(socketTimeoutStr);

                if (configValue > 0) {
                    defaultSocketTimeout = configValue;
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
            socketTimeoutNum = defaultSocketTimeout;
        }
        return socketTimeoutNum.intValue();
    }
}

/**
     * 根据属性key得到对应的属性值<br>
     * 配置文件按spring boot的约定的配置文件命名规则进行加载<br>
     * <font color=red>注意： 暂不支持yml文件属性读取</font>
     * 
     * @param key 键名称
     * @return
     */
    public static String getPropsValueByKey(String key) {
        // TODO 暂时不支持读取yml格式文件，yml文件支持map和list格式数据，需要另写方法支持
        if (!props.containsKey(CONFIG_FILENAME)) {
            Properties prop = new Properties();
            prop = getPropertiesByFileName(CONFIG_FILENAME);

            if (prop.get(SPRING_PROFILES_ACTIVE) != null) {
                // 依次读取指定的配置文件
                for (String partName : prop.get(SPRING_PROFILES_ACTIVE).toString().split(",")) {
                    prop.putAll(getPropertiesByFileName(SPRING_BOOT_PROFILE_TEMPLATE.replace("{profile}", partName)));
                }
            }
            props.put(CONFIG_FILENAME, prop);
        }
        Object obj = props.get(CONFIG_FILENAME).get(key);
        if (obj == null) {
            return null;
        } else {
            return obj.toString();
        }
    }

作者 east

Spark 3月 16,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

分类归档大数据开发