linux操作系统是典型的机制和策略相分离的系统，机制和策略一直是设计领域一直关注的重要问题，它们的结合程度直接关系到系统的稳定性和可扩展性。今天做了一个linux上膝上电脑的电源管理的脚本，颇有感触，故写此文。   
对于膝上电脑，最大的特点就是靠电池供电，而电脑最耗电的就是机械装置，比如磁盘，因此如何使得磁盘效率最高就是操作系统的任务了，磁盘调度算法当然是最重要的，可是本文不讨论它们。操作系统如果想提供省电模式，可以直接在内核实现一系列的策略，然后用户选择膝上电脑模式或者系统自动监测到电池的时候切换 到相应的磁盘模式，这个实现看似职能，实际上很不好，极大的限制了用户。linux的策略和机制是分离的，因此，体现到这个事情上也是如此。    
     linux内核有着非常强大的磁盘缓存机制，就是磁盘数据先不往磁盘直接读写而是直接操作缓存，待到一定条件满足的时候才读写磁盘，大致有几个参数：1.dirty_writeback_centisecs，这个参数表示内核刷新缓存的时间间 隔；2.dirty_expire_centisecs，这个参数表示一个inode在dirty状态停留的最长时间；3.laptop_mode，这个 参数表示是否为膝上电脑模式。系统根据以上参数动态调整磁盘刷新的行为，如果为膝上电脑模式，那么我们需要让磁盘尽量少的走走停停，就是说将脏缓存页面积累到一定程度，然后一次性写入磁盘而不是像pc机那样小间隔的唤醒pdflush，这样的话可以让磁盘的效率更高，因此我们理所当然的要在 laptop_mode参数为1的时候调整上面提到的另外两个参数，很早以前我想当然的以为内核在监测到laptop_mode为1时会动态调整那两个参 数，可是我没有找到代码，当我知道linux的机制和策略相分离的时候，我才明白，linux内核是不管那种事的，如果管的话，它怎么管，它知道调整到具体哪个值吗？如果按照所有的可能来个switch-case的话，内核的大小就要增加好几倍了，windows一直都以大而著称，就是因为它的内核实现好 多的策略，而这些策略是可以在用户空间微调的，比如，针对上述问题我可以写一个脚本：    
...    
if [ ! -w /proc/sys/vm/laptop_mode ]; then    
    exit 1    
fi    
case "$1" in     
    laptop)         
        echo "1"    > /proc/sys/vm/laptop_mode    
        echo 参数1    > /proc/sys/vm/dirty_writeback_centisecs    
        echo 参数2    > /proc/sys/vm/dirty_expire_centisecs    
        echo 参数3    > /proc/sys/vm/dirty_ratio    
        echo 参数4    > /proc/sys/vm/dirty_background_ratio    
        ...    
    non-laptop)    
        echo "0"    > /proc/sys/vm/laptop_mode    
        echo 参数1-1    > /proc/sys/vm/dirty_writeback_centisecs    
        echo 参数2-2    > /proc/sys/vm/dirty_expire_centisecs    
        echo 参数3-3    > /proc/sys/vm/dirty_ratio    
        echo 参数4-4    > /proc/sys/vm/dirty_background_ratio    
        ...    
这 里的“参数X”是可以微调的，看到了吧，多么灵活！linux就像一个布满微调按钮的机器，或者是一个按键十分丰富的控制台，初接触的人往往望而生畏，因 为它太复杂了，几乎什么都可以微调，可以做成任何你需要的系统，而不像别的操作系统，比如windows，什么都为你做好，你只要选择就可以了，是的，选 择模式固然很方便，但是却十分不灵活，很多功能你要想实现要费尽周折，这就是为何windows非常大，功能却很有限，linux小，功能却很丰富的原 因。

前言：
基于logstash-input-jdbc较其他插件的稳定性、易用性、版本和ES同步更新的特点，以下研究主要针对 logstash-input-jdbc 展开。 
针对logstash-input-jdbc常见的几个疑难问题，部分问题也在git和stackoverflow进行了激烈讨论，以下统一给出验证和解答。
1、logstash-input-jdbc 的同步原理是什么？
（1）、对于全量同步依据
配置文件jdbc.sql的sql语句的进行同步。
（2）、对于增量实时同步依据
1）设定的定时策略。
如最小更新间隔每分钟更新一次设定：schedule => “* * * * *”，目前最小更新间隔为1分钟，验证发现，不支持60s以内的秒级更新。
2）设定的sql语句。
如jdbc.sql, 决定同步哪些内容及同步更新的条件。
{"id":10,"name":"10test","@version":"1","@timestamp":"2016-06-29T03:18:00.177Z","type":"132c_type"}
1
2：logstash-input-jdbc 只支持基于时间同步吗？
验证表名：同步更新除了支持根据时间同步外，还支持根据某自增列（如：自增ID）字段的变化进行同步。
上次举例只是举了同步时间变化的例子，设定条件：
[root@5b9dbaaa148a logstash_jdbc_test]# cat jdbc.sql_bak

select
        *
from
        cc
where   cc.modified_at > :sql_last_value
1
2
3
4
5
6
7
实际进一步研究发现，在配置文件中有use_column_value字段决定，是否需要记录某个column 的值,如果 record_last_run 为真,可以自定义我们需要 track 的 column 名称，此时该参数就要为 true. 否则默认 track 的是 timestamp 的值.
举例：以下即是设定以id的变化作为同步条件的。
[root@5b9dbaaa148a logstash_jdbc_test]# cat jdbc_xm.sql
select
        *
from
        cc
where   cc.id >= :sql_last_value
1
2
3
4
5
6
我们可以指定文件,来记录上次执行到的 tracking_column 字段的值 比如上次数据库有 12 条记录,查询完后该文件中就会有数字 12 这样的记录,下次执行 SQL 查询可以从 13 条处开始.
我们只需要在 SQL 语句中 WHERE MY_ID > :last_sql_value 即可. 其中 :last_sql_value 取得就是该文件中的值(12).
last_run_metadata_path => “/etc/logstash/run_metadata.d/my_info”
如：
[root@5b9 run_metadata.d]# cat /etc/logstash/run_metadata.d/my_info

--- 12
1
2
3
已全局代码搜索，没有触发器trigger相关处理操作。
3：mysql和ES分别存储在两台服务器，且时间不一致，能否实现同步？
（1）. 设定对于以时间作为判定条件的增量同步，可以以设定的时间为基准点进行同步。
验证发现：
显示的时间戳timestamp为ES上的UTC时间值（不论ES机器是什么时区，都会修改为UTC时间存入ES），显示的modified_at时间值为同步过来的mysql时间值转化为UTC的结果值。
更新的前提是必须满足： cc.modified_at >= :sql_last_value。即如果mysql的时间修改为小于sql_last_value的时刻值，是无法进行同步的。
   如：
[elasticsearch@5b9dbaaa148a run_metadata.d]$ cat my_info

--- 2016-06-29 02:19:00.182000000 Z
1
2
3
（2）. 对于选定某列作为判定条件（如自增ID），两者（mysql和ES)时间不一致，实际是也可以同步更新的。
验证发现：
测试设定的时间是mysql比ES早一天或者晚一天的时刻值，都可以实现同步更新操作。
4：如何支持实时同步mysql的delete操作到ES中？
logstash-input-jdbc插件不支持物理删除的同步更新。详见：
http://stackoverflow.com/questions/35813923/sync-postgresql-data-with-elasticsearch/35823497#35823497
https://github.com/logstash-plugins/logstash-input-jdbc/issues/145
解决方案：
同步删除操作改为同步update更新操作实现。
第一步：进行软件删除，而不是物理删除操作。
先不物理删除记录，而是软件删除，即新增一个 flag 列，标识记录是否已经被删除（默认为false，设置为true或者deleted代表已经被删除，业界通用方法），这样，通过已有的同步机制，相同的标记记录该行数据会同步更新到Elasticsearch。
第二步：ES中检索flag标记为true或者deleted的字段信息。
在ES可以执行简单的term查询操作,检索出已经删除的数据信息。
第三步：定时物理删除。
设置定时事件，间隔一段时间物理删除掉mysql和ES中的flag字段标记为true或deleted的记录，即完成物理删除操。

本文源码：GitHub·点这里 ||  GitEE·点这里
一、DataX工具简介
1、设计理念
DataX是一个异构数据源离线同步工具，致力于实现包括关系型数据库(MySQL、Oracle等)、HDFS、Hive、ODPS、HBase、FTP等各种异构数据源之间稳定高效的数据同步功能。解决异构数据源同步问题，DataX将复杂的网状的同步链路变成了星型数据链路，DataX作为中间传输载体负责连接各种数据源。当需要接入一个新的数据源的时候，只需要将此数据源对接到DataX，便能跟已有的数据源做到无缝数据同步。
絮叨一句：异构数据源指，为了处理不同种类的业务，使用不同的数据库系统存储数据。
2、组件结构
DataX本身作为离线数据同步框架，采用Framework+plugin架构构建。将数据源读取和写入抽象成为Reader和Writer插件，纳入到整个同步框架中。
Reader
Reader为数据采集模块，负责读取采集数据源的数据，将数据发送给Framework。
Writer
Writer为数据写入模块，负责不断向Framework取数据，并将数据写入到目的端。
Framework
Framework用于连接reader和writer，作为两者的数据传输通道，并处理缓冲，流控，并发，数据转换等核心技术问题。
3、架构设计
Job
DataX完成单个数据同步的作业，称为Job，DataX接受到一个Job之后，将启动一个进程来完成整个作业同步过程。Job模块是单个作业的中枢管理节点，承担了数据清理、子任务切分(将单一作业计算转化为多个子Task)、TaskGroup管理等功能。
Split
DataXJob启动后，会根据不同的源端切分策略，将Job切分成多个小的Task(子任务)，以便于并发执行。Task便是DataX作业的最小单元，每一个Task都会负责一部分数据的同步工作。
Scheduler
切分多个Task之后，Job会调用Scheduler模块，根据配置的并发数据量，将拆分成的Task重新组合，组装成TaskGroup(任务组)。
TaskGroup
每一个TaskGroup负责以一定的并发运行完毕分配好的所有Task，默认单个任务组的并发数量为5。每一个Task都由TaskGroup负责启动，Task启动后，会固定启动Reader—>Channel—>Writer的线程来完成任务同步工作。DataX作业运行起来之后，Job监控并等待多个TaskGroup模块任务完成，等待所有TaskGroup任务完成后Job成功退出。否则，异常退出，进程退出值非0。
二、环境安装
推荐Python2.6+，Jdk1.8+(脑补安装流程)。
1、Python包下载
# yum -y install wget
# wget https://www.python.org/ftp/python/2.7.15/Python-2.7.15.tgz
# tar -zxvf Python-2.7.15.tgz
2、安装Python
# yum install gcc openssl-devel bzip2-devel
[root@ctvm01 Python-2.7.15]# ./configure --enable-optimizations
# make altinstall
# python -V
3、DataX安装
# pwd
/opt/module
# ll
datax
# cd /opt/module/datax/bin
-- 测试环境是否正确
# python datax.py /opt/module/datax/job/job.json
三、同步任务
1、同步表创建
-- PostgreSQL
CREATE TABLE sync_user (
id INT NOT NULL,
user_name VARCHAR (32) NOT NULL,
user_age int4 NOT NULL,
CONSTRAINT "sync_user_pkey" PRIMARY KEY ("id")
);
CREATE TABLE data_user (
id INT NOT NULL,
user_name VARCHAR (32) NOT NULL,
user_age int4 NOT NULL,
CONSTRAINT "sync_user_pkey" PRIMARY KEY ("id")
);
2、编写任务脚本
[root@ctvm01 job]# pwd
/opt/module/datax/job
[root@ctvm01 job]# vim postgresql_job.json
3、脚本内容
{
"job": {
"setting": {
"speed": {
"channel": "3"
}
},
"content": [
{
"reader": {
"name": "postgresqlreader",
"parameter": {
"username": "root01",
"password": "123456",
"column": ["id","user_name","user_age"],
"connection": [
{
"jdbcUrl": ["jdbc:postgresql://192.168.72.131:5432/db_01"],
"table": ["data_user"]
}
]
}
},
"writer": {
"name": "postgresqlwriter",
"parameter": {
"username": "root01",
"password": "123456",
"column": ["id","user_name","user_age"],
"connection": [
{
"jdbcUrl": "jdbc:postgresql://192.168.72.131:5432/db_01",
"table": ["sync_user"]
}
],
"postSql": [],
"preSql": []
}
}
}
]
}
}
4、执行脚本
# /opt/module/datax/bin/datax.py /opt/module/datax/job/postgresql_job.json
5、执行日志
2020-04-23 18:25:33.404 [job-0] INFO  JobContainer -
任务启动时刻                    : 2020-04-23 18:25:22
任务结束时刻                    : 2020-04-23 18:25:33
任务总计耗时                    :                 10s
任务平均流量                    :                1B/s
记录写入速度                    :              0rec/s
读出记录总数                    :                   2
读写失败总数                    :                   0
四、源码流程分析
注意：这里源码只贴出核心流程，如果要看完整源码，可以自行从Git上下载。
1、读取数据
核心入口：PostgresqlReader
启动读任务
public static class Task extends Reader.Task {
@Override
public void startRead(RecordSender recordSender) {
int fetchSize = this.readerSliceConfig.getInt(com.alibaba.datax.plugin.rdbms.reader.Constant.FETCH_SIZE);
this.commonRdbmsReaderSlave.startRead(this.readerSliceConfig, recordSender,
super.getTaskPluginCollector(), fetchSize);
}
}
读取任务启动之后，执行读取数据操作。
核心类：CommonRdbmsReader
public void startRead(Configuration readerSliceConfig,
RecordSender recordSender,
TaskPluginCollector taskPluginCollector, int fetchSize) {
ResultSet rs = null;
try {
// 数据读取
rs = DBUtil.query(conn, querySql, fetchSize);
queryPerfRecord.end();
ResultSetMetaData metaData = rs.getMetaData();
columnNumber = metaData.getColumnCount();
PerfRecord allResultPerfRecord = new PerfRecord(taskGroupId, taskId, PerfRecord.PHASE.RESULT_NEXT_ALL);
allResultPerfRecord.start();
long rsNextUsedTime = 0;
long lastTime = System.nanoTime();
// 数据传输至交换区
while (rs.next()) {
rsNextUsedTime += (System.nanoTime() - lastTime);
this.transportOneRecord(recordSender, rs,metaData, columnNumber, mandatoryEncoding, taskPluginCollector);
lastTime = System.nanoTime();
}
allResultPerfRecord.end(rsNextUsedTime);
}catch (Exception e) {
throw RdbmsException.asQueryException(this.dataBaseType, e, querySql, table, username);
} finally {
DBUtil.closeDBResources(null, conn);
}
}
2、数据传输
核心接口：RecordSender(发送)
public interface RecordSender {
public Record createRecord();
public void sendToWriter(Record record);
public void flush();
public void terminate();
public void shutdown();
}
核心接口：RecordReceiver(接收)
public interface RecordReceiver {
public Record getFromReader();
public void shutdown();
}
核心类：BufferedRecordExchanger
class BufferedRecordExchanger implements RecordSender, RecordReceiver
3、写入数据
核心入口：PostgresqlWriter
启动写任务
public static class Task extends Writer.Task {
public void startWrite(RecordReceiver recordReceiver) {
this.commonRdbmsWriterSlave.startWrite(recordReceiver, this.writerSliceConfig, super.getTaskPluginCollector());
}
}
写数据任务启动之后，执行数据写入操作。
核心类：CommonRdbmsWriter
public void startWriteWithConnection(RecordReceiver recordReceiver,
Connection connection) {
// 写数据库的SQL语句
calcWriteRecordSql();
List
关于找一找教程网
本站文章仅代表作者观点，不代表本站立场，所有文章非营利性免费分享。
本站提供了软件编程、网站开发技术、服务器运维、人工智能等等IT技术文章，希望广大程序员努力学习，让我们用科技改变世界。
[数据源管理 | 基于DataX组件，同步数据和源码分析]http://www.zyiz.net/tech/detail-132867.html

1.关系型数据库和非关系型数据库的区别（优缺点，适用环境）

2.博客为什么是mongodb

1.关系型

代表数据库系统：MySQL， SQL Server， Oracle，

关系型数据库的表是二维表，结构稳定，修改不易。经常需要联表查询。

优点：

1.查询能力高，可以操作很复杂的查询

2.一致性高。由于并发高，在数据同步的时候一般采用锁来保证数据的可靠性（还有时间戳什么的方法），在处理数据的时候对表进行封锁来保证操作的时候其他的操作不能改变当前查询范围内的数据的值。

3.由于表具有逻辑性，易于理解。

既然它现在有些过气了（2333）自然有原因，随着大数据时代的到来，关系型数据库的严谨变得有些不适用：

1.不适用高并发读写

2.不适用海量数据高效读写

3.层次多，扩展性低

4.维护一致性开销大

5.涉及联表查询，复杂，慢

2.非关系型

非关系型数据库主要是使用key-value的方式存储数据的，即是『no relational』，这是『显然』区别于关系型数据库的特点之一。由于数据间没有关联性，相对来说层级扁平。

优点：

1.由于数据之间没有关系，所以易扩展，也易于查询

2.数据结构灵活，每个数据都可以有不同的结构

3.由于降低了一致性的要求，所以查询速度更快

非关系型数据库的产生是因为随着网站的进化，并发性增加，扩展性高，一致性要求降低。这样关系型数据库最重要的一致性维护就显得有点多余，并且消耗着性能。因此有了非关系型数据库，它可以算是关系型数据库的一种弱化的结果，在海量数据存储和查询上更胜一筹。

两种数据库没有好坏之分，只是使用的环境不一样。关系型数据库可以说是更严谨的，可靠性更强的数据库，在对于数据精度要求高的环境，比如说银行系统这样自然是像mysql这样的数据库适合。非关系型数据库胜在处理大数据的速度，但是对于数据的准确度没有那么高，对于操作量大的环境比如当前大部分web2.0的网站更加适用一些。

3.Why mongodb

主要比较下mongodb和redis。

『

MongoDB更类似MySQL，支持字段索引、游标操作，其优势在于查询功能比较强大，擅长查询JSON数据，能存储海量数据，但是不支持事务。

Mysql在大数据量时效率显著下降，MongoDB更多时候作为关系数据库的一种替代。

mongodb更吃内存，因为当mongo发现内存不够的时候，是以2的指数级别来申请内存的。所以一般都建议把mongodb单独放。

其实可以说redis更像缓存机制，cookie，也可以设定数据的过期时间，当然也可以永久存储（但是好像稍逊色？）。mongodb是文档式的存储，在类似于博客系统这样，可以把每篇博客，包括日期，评论什么的都当做一篇文档来看。这样在博客数据库选择的时候，自然是mongodb更加贴近一些（如果选择非关系型数据库）， 也相较于redis更适合永久存储

内存管理机制

Redis数据全部存在内存，定期写入磁盘，当内存不够时，可以选择指定的LRU算法删除数据。

MongoDB和mysql一样，只是把索引文件放到内存中。由linux系统mmap实现，当内存不够时，只将热点数据放入内存，其他数据存在磁盘。

（注：mmap （一种内存映射文件的方法） mmap将一个文件或者其它对象映射进内存。文件被映射到多个页上，如果文件的大小不是所有页的大小之和，最后一个页不被使用的空间将会清零。mmap在用户空间映射调用系统中作用很大。）

支持的数据结构

Redis支持的数据结构丰富，包括hash、set、list等。

MongoDB数据结构比较单一，但是支持丰富的数据表达，索引，最类似关系型数据库，支持的查询语言非常丰富。

性能

redis更适用于较小数据量的性能及运算

mongodb则在海量数据的访问下性能更优

可靠性

二者均支持持久化。

集群

MongoDB集群技术比较成熟，Redis从3.0开始支持集群。

不适用场景

Ø 需要使用复杂sql的操作

Ø 事务性系统

』

转载

原文

Redis 最佳应用场景：适用于数据变化快且数据库大小可遇见（适合内存容量）的应用程序。

例如：股票价格、数据分析、实时数据搜集、实时通讯。

MongoDB：最佳应用场景：适用于需要动态查询支持；需要使用索引而不是 map/reduce功能；需要对大数据库有性能要求；需要使用 CouchDB但因为数据改变太频繁而占满内存的应用程序。

例如：你本打算采用 MySQL或 PostgreSQL，但因为它们本身自带的预定义栏让你望而却步。

简说mmap:

mmap系统调用并不是完全为了用于共享内存而设计的。它本身提供了不同于一般对普通文件的访问方式，进程可以像读写内存一样对普通文件进行操作。

mmap 系统调用使得进程之间通过映射同一个普通文件实现共享内存。普通文件被映射到进程地址空间后，进程可以像访问普通内存一样对文件进行访问，不必再调用。 read()，write（）等操作。mmap并不分配空间, 只是将文件映射到调用进程的地址空间里, 然后你就可以用memcpy(内存拷贝)等操作写文件, 而不用write()了.写完后用msync()同步一下（msync()：一般说来，进程在映射空间的对共享内容的改变并不直接写回到磁盘文件中，往往在调用munmap（解除映射关系）后才执行该操作。可以通过调用msync()实现磁盘上文件内容与共享内存区的内容一致）, 你所写的内容就保存到文件里了. 不过这种方式没办法增加文件的长度, 因为要映射的长度在调用mmap()的时候就决定了。

大数据处理的模型：MapReduce

Mapreduce是一个分布式运算程序的编程框架，是用户开发“基于hadoop的数据分析应用”的核心框架；

Mapreduce核心功能是将用户编写的业务逻辑代码和自带默认组件整合成一个完整的分布式运算程序，并发运行在一个hadoop集群上

MapReduce的处理过程分为两个步骤：map和reduce。map和reduce函数是要执行的任务，由master分配任务给worker执行。map函数读取被分配的输入数据片段，输出中间key/value pair值的集合，reduce函数收集具有相同中间key值的value值，合并这些value值，形成一个较小的value值的集合。

MapReduce详解

从零开始学Hadoop——浅析MapReduce（一）

---------------------
https://blog.csdn.net/qq_33915826/article/details/80386321