一、项目背景介绍

经过十三年的发展，快递100目前C端累计注册用户超2.5亿、P端（专业用户）累计注册快递员及网点经营者超130万、B端累计服务电商企业/泛电商企业/品牌企业/政府与公共组织等客户超250万家；每天快递查询调用量超4亿次、寄件下单量超30万单。

公司的业务量和数据量是相对较大且复杂的，因此拥有一个实时性、可扩展性、并拥有强大的搜索与分析功能的日志中心至关重要，它不仅可以记录系统的性能、运行状态，还可以为我们提供很多有价值的业务数据和用户行为分析，这些都将为业务的洞察与决策提供有效支持，从而推动产品的迭代升级和运营策略调整。

同时，日志也是我们线上定位问题的重要手段，我们经常会依赖日志排查解决来自客户的问题，帮助我们提升服务质量，增强客户对公司的产品信赖。

如何构建一套适合我们的日志中心？在业务的发展过程中，日志的架构从原始的文件记录到ELK体系，我们也遇到了一系列的问题，经历实践和研究分析，我们最终构建了新一代的日志系统。下面就给大家分享整个过程。

二、初期架构1. 原始架构

把时间回溯到10年前，我们会怎样去记录日志？

架构比较简单，但问题也比较明显，有几个明显的缺点：

• 每次查看日志文件都需要登录到不同的机器，非常不方便；
• 通过 tail 或者 cat 等命令查看日志，如果对日志文件进行检索、聚合等操作，还会对服务器的 io 造成很大的压力，甚至导致故障的产生；
  
• 日志文件过大不仅导致查询变得特别慢，还经常带来磁盘告警甚至磁盘空间不足等严重后果；
  
• 日志格式不规范，日志随意写到文件，可读性和可分析性几乎为零；
• 应用多节点挂载 NFS 性能差，容易产生日志丢失，从而影响问题定位和排障。
  

2.ELK体系

经历了这些问题之后，2017年，我们把目光转向新一代的基于Elasticsearch的日志体系，恰逢之前快递100订单检索引入了Elasticsearch，侧面又了解到基于Elastic Stack的ELK体系，经过一系列研究，开始意识到日志规范的重要性和采集的好处。

ELK是业界最成熟的日志技术栈，使用JSON格式存储，易于解析，再配合全文检索能力，能够快速从众多的日志信息中搜索到关键信息，加上Kibana的易用性，使得日志体验上升了几个档次，架构大致如下，不具体拆分细讲。

我们再回顾一下之前的挑战和缺点，可以看到基本解决了之前遇到的问题：

三、实践中的挑战

使用ELK一段时间之后，我们解决了从无到有的问题。但随着一系列基于日志进行分析和告警的工作逐步开展，新的问题也开始浮现。我们尝试进行优化后，效果并不明显，这促使我们重新考量架构的升级，主要的问题体现在：

• 成本问题：ES压缩率不高，基于目前的日志量，合规性要求需要保留6个月，需要耗费巨大的存储成本；
• 吞吐瓶颈：ES分词特性写入吞吐瓶颈问题，容易导致日志写入发生延迟；
• 资源占用率高：ES在内存使用上的消耗过高；
• 生命周期维护：ES旧版本TTL问题，需要手工介入数据过期，维护成本高；
• 分析能力一般：由于更多的分析需求出现，ES的聚合能力受到了挑战。

针对上述的一些问题，在2020年开始了解到Clickhouse的存在，我们对ES和Clickhouse做了一个选型对比。基于对比得出的结论，最终决定选用 Clickhouse 作为下一代日志存储数据库。

四、新的架构体系

ELK体系经过多年的发展，生态已经非常强大，Clickhouse想达到同样的生态，需要更长的时间去发展，因此这个过程也需要投入一些研究或者开发量才能达到更好的效果，幸好 Clickhouse 本身的学习曲线较低，经过短时间的研究，我们制定了新的日志架构。

可以从几个组件的构成来看这个架构图，对比ELK体系与新的体系的不同：

• 采集层：从Logstash到ilogtail，ilogtail性能更强，资源消耗更低；
• 处理层：从Logstash到ilogtail，ilogtail还支持数据脱敏，多行拆分等实用功能；
• 存储层：从Elasticsearch改为Clickhouse，选型过程已有对比，这里不再赘述；
• 可视化：从Kibana到Clickvisual，这里有优点也有缺点，所以还是配合了Grafana才达到类似的效果。

Clickvisual的优点：灵活的SQL、日志审计、告警策略；

Kibana的优点：Kibana具备一些基础的BI功能，可用于日志分析。

1.新架构的成果

还是回顾一下之前的挑战，问题基本得到了解决：

2.基于Clickhouse的日志存储

基于10亿日志数据进行测试，得出磁盘占用的对比柱状图：

基于10亿数据的测试，在两者集群模式下，消费Kafka的速度对比：

新的架构最核心的改变，就是将ES换成了Clickhouse，看中的就是极高的压缩率，最终的结果是同等存储条件下，原来ES只能保留一个月的数据，现在可以做到保留六个月，这其中少不了很多存储细节的优化，其中包含：

• 大部分字段采用ZSTD压缩模式来提升压缩率；
• 低基数LowCardinality的使用，节省存储的同时还做到性能提升；
• 连续性时间字段的Delta+ZSTD压缩；
• 冷热策略的配置，近一个月保留在SSD盘，一到六个月的数据自动流到HDD盘，六个月前数据自动清理。

建表语句如下：

3.基于Clickvisual的可视化

ClickVisual 是一个轻量级的开源日志查询、分析、报警的可视化平台，致力于提供一站式应用可靠性的可视化的解决方案。既可以独立部署使用，也可作为插件集成到第三方系统。目前是市面上唯一一款支持 ClickHouse 的类 Kibana 的开源业务日志查询平台。

它具备的特性，部分符合我们的需求：

• 支持可视化的查询面板，可查询命中条数直方图和原始日志；
• 支持设置日志索引功能，分析不同索引的占比情况；
  
• 支持 Proxy Auth 功能，能被非常轻松地集成到第三方系统；
  
• 支持基于 ClickHouse 日志的实时报警功能。
  

还提供了原始的SQL查询功能，直接输入SQL聚合语句，即时简单地对日志进行聚合分析：

体验总体类似Kibana，细节稍有不足，通过这个查询分析界面作为一个入口，搭配日志告警模块，快速定位问题和故障排除方面的能力得到了大大的提升，基本无缝从Kibana上切换过来，拥有不错的排障体验。

五、进一步优化

诚然，做到上述的效果还是不足以满足我们的要求。因而在此基础上，我们进行了优化方面的思考 ，其中也踩了一些Clickhouse的坑，使用了一些Clickhouse的新特性，是一个很有意思的过程。1.日志查询优化探索

得益于Clickhouse的高压缩率和查询性能，小日志量的表直接配合时间分区搜索即可，但是当日志量涨到一定程度的时候，查询缓慢总是一个难受的事，我们对一些场景进行了总结：

• traceid场景：在 Skywalking 中根据 traceid 查询链路日志时，使用 tokenbf_v1索引，并通过 hasToken 查询，由于跳过大部分无效 parts，可快速命中返回；

• 无结构化日志：对于这种无结构化日志，用 like 性能会非常慢且消耗 CPU 甚至内存，新版本的 Clickhouse 已经支持了倒排索引，也开始基于倒排— 索引优化，可大大提高响应速度；

• 聚合场景：一些常规的聚合需求，可通过Clickhouse的Projection功能来满足。

2.本地表还是分布式表

本地表是Clickhouse的存储表，分布式表只是逻辑表，本身并不存储数据，在日志高频写入的场景，还是推荐写本地表，原因有这么几点：

• 当我们大批量写入日志时，可以直接往分布式表写，但数据会先拆分成不同parts，再通过Zookeeper进行分发，增加了集群间网络的负载，导致写入变慢，甚至出现Too many parts问题；
• 写分布式表更容易出现数据一致性问题；
• Zookeeper压力变大。

3.Clickhouse的限制策略

随着日志中心的建设，日志体量越来越大，开始暴露一些配置层面的问题，我们开始对 Clickhouse 增加一些限制，以免错误的 SQL 导致集群缓慢甚至 OOM 的问题，对于日志查询用户，单独做了 SQL 复杂度的限制，users.xml 中有几个参数：

• max_memory_usage：单个服务器上运行查询的最大内存；
• max_memory_usage_for_user：单个服务器上运行用户查询的最大内存；
• max_memory_usage_for_all_queries：单个服务器上运行所有查询的最大内存；
• max_rows_to_read：运行查询时可从表中读取的最大行数；
• max_result_rows：限制结果中的行数；
• max_bytes_to_read：运行查询时可以从表中读取的最大字节数（未压缩数据）。

结语

在这篇文章中，我们分享了快递100在云原生技术方面的实践、思考和应用，特别是在日志中心建设方面的实践。日志中心的上线应用，让我们在问题定位方面的效率得到了极大的提升，系统更加稳定和可靠；同时，通过对日志的收集、分析和挖掘，我们也更好地了解了用户需求和行为，优化了产品设计和运营策略，促进业务高速增长。

from:https://mp.weixin.qq.com/s?__biz=MzkzMjYzNjkzNw==&mid=2247612190&idx=1&sn=046e724f035dedd610ffb43e56ae07ea&source=41#wechat_redirect

建设实时数仓的目的和意义

实时数仓目的

数仓概念：数据尽可能多，保存时间尽可能久

实时概念：数据流式，处理及时、瞬时、短时、事件或者微批响应

数仓跟实时从概念上就有冲突，所以本质上不太适合处理广泛的问题，比如，对一个月，甚至是一年的数据进行统计计算。

所以，实时数仓应该目前作为离线数仓的一种补充，解决因离线数仓实时性低而无法解决的问题，具体点说就是处理离线两个周期间隔的数据问题，不适合解决大批量数据聚合问题、业务性太强的以及对实时性要求很高问题。

实时数仓的意义

实时数仓从概念上讲还是要靠近数仓的概念，数据分层，面向主题，数据尽可能集成，结构相对稳定，不易发生变化。

对于实时数仓来讲，数据量不需要保存像离线那么久，上一节我们提到，实时数仓处理两个离线周期间隔的数据即可，如上图，以时报为例，实时数仓补充中间数据即可，以天为例，实时数仓最多只需要保留3~5天数据即可，能够支持一段时间的数据追溯和重导就可以了。

实时数仓可以解决哪类问题

利用EMR建设实时数仓

实时数仓对比离线数仓

实时数仓架构

从图中可以看到，

1、ODS并不是实时数仓的一部分，是依赖外部数据源，比如binlog,流量日志，系统日志，或者是其他消息队列

2、应用层也不是实时数仓的一部分，对于数据的使用，通过实时数仓暴露Topic来使用

3、实时数仓要求层次要少，因为需要尽可能降低延迟

用EMR搭建实时数仓

1、底层数据源可以接企业内部binlog、日志或者消息队列

2、从ODS层经过与维表轻度扩展，形成明细层明细表，明细表用一个Ckafka topic表示，计算采用Oceanus或者EMR FlinkSql 关联查询，维表采用EMR Hbase存储

3、从明细层经过进一步汇总计算，形成汇总层，此时数据已经是面向主题的汇总数据，就是传统意义上的大宽表，一个主题是一系列Ckafka topic，计算采用Oceanus或者EMR FlinkSql 关联查询以及汇总计算

实时数仓各层搭建

ODS层搭建

1、 之所以没有把ODS层放在实时数仓的一部分，是因为实时数仓的ODS并不像离线数仓ODS是采集过来的原始数据，现在一般企业都已经具备了如上图的底层数据源

2、 Binlog，是数据库日志，通过Binlog可以自数据库主从间同步，可以同步关系型数据库数据，目前企业线上数据库都采用Mysql这样的数据库，可以通过抓取Mysql binlog 获取数据库变更信息，数仓中重要的业务数据，支付相关，用户相关，管理相关数据一般都从这种数据源获得

3、 Log日志，服务器日志，像服务器系统日志采集，都是通过这种形式进行采集

Ckafka，企业通过消息队列提供数据源服务，比如，点击流服务，会把用户点击事件通过上报服务器上报到Ckafka，为后续分析提供原始数据

该层搭建的注意点：
1、 业务选择数据源，尽量跟离线保持一致，比如某个业务，数据源即可以通过Binlog，也可以通过Log日志采集，如果离线数仓业务是通过Binlog，那么实时数仓也取Binlog，否则后续产生数据不一致，非常难以定位

2、数据源要求一致性，对于Ckafka和Binlog 需要进行分区一致性保证，解决数据乱序问题

明细层搭建

建设标准与离线数仓目标一致，解决原始数据存在噪音，不完整，形式不统一等问题

数据解析，业务整合，数据清洗，解决噪音，不完整，数据不一致问题;模型规范化（提前指定号规则，尽量跟离线保持一致），形成数据规范，规范尽可能跟离线保持一致，命名，比如，指标命名等;

与离线数仓不同之处在于，离线调度是有周期的，时报一小时，天报周期为一天，如果修改数据表字段，只要任务没开始，就可以修改，而实时是流式，7X24小时不间断运行的，想要修改流中的字段或者格式，对下游影响是不可预估的

实时数仓如果修改字段不像离线，在间隔期间通知下游把作业都改了就没事了，但是实时不一样，实时你改掉了字段，下游作业必须可以认识你修改的内容才行，kafka不是结构化存储，没有元数据的概念，不像Hive，如果表名不规范，找一个统一时间，把catolog改规范，然后把脚本一改就就解决了。

明细建设关键，我们会在每一条数据上增加一些额外字段到数仓里

举例说明这些额外字段的意义

事件主键：对于上游数据重复问题，我们会根据一些数据内的字段来判断上有数据的唯一性，比如binlog，<集群id_><库id_><表id_>数据id_数据生成时间。

数据主键：唯一标识数据表的一行记录，可以使用数据库主键，主要用来解决分区一致性及分区有序。

数据元数据版本：上面介绍了，流式计算是7X24小时不间断的运算，当修改了数据结构，增加，删除了字段，对下游的影响是不可预估的，因此元数据变更需修改该字段，保持数据流中新老版本数据双跑，下游选择合适的时机进行数据切换。

数据批次：跟元数据用途相似，当明细层逻辑发现问题，需要重跑数据，为了对下游任务不产生影响，调整了明细层逻辑后，需要回倒位点重跑数据，同时需要跟老逻辑任务双跑，待下游业务都切换到新的逻辑后，老逻辑任务才可以停止。

还有一个思路，可以直接把明细层数据，也可以直接写到druid 直接用于分析。

维度层搭建

维度数据处理：

如上图，对于变化频率低，地理，节假日，代码转换，直接同步加载到缓存里，或者是新增数据，但是增加进来就不变了，通过数据接口，访问最新数据，然后通过本公司数据服务对外提供数据

如上图，对于变化频率高，比如商品价格，也是需要监听变化消息，然后实时更新维度拉链表。对于比如像最近一个月没有消费用户这样的衍生维度，是需要根据变化消息，通过计算得到的衍生维度拉链。

因为维度数据也在发生变化，为了能够让源表数据匹配到维表，我们会给维表增加多版本minversion，然后通过TIMERANGE => [1303668804,
1303668904]筛选出源数据指定的维表版本数据。

这里有些同学可能觉得如果版本一致保存下去，会不会非常大，是的，我们响应的需要配置TTL保证维表数据量可控，上文我们介绍过，实时数仓解决是离线数仓两个间隔的问题，那么像这种变化频繁的数据我们TTL设置一周足够了。

关于源表与维表如果进行join，Flink原生sql以及Oceanus都是采用UDTF函数以及Lateral
Table 进行联合使用，其中UDTF我们可以实现查询数据服务获取维表数据的能力，Oceanus请参考相关材料。

汇总层搭建

汇总层加工其实跟离线数仓是一致的，对共性指标进行加工，比如，pv，uv，订单优惠金额，交易额等，会在汇总层进行统一计算。

Flink提供了丰富的窗口计算，这使得我们可以做更细力度的聚合运算，例如，我们可以算最近5分钟，10分钟的数据聚合，根据时间窗口的间隔，也需要调整相应的TTL，保障内存高效实用。

Flink提供了丰富的聚合计算，数据都是要存在内存中的，因此需要注意设置state的TTL，例如，做Count（Distinct
x）。或者在进行PV，UV计算时候，都会使用大量的内存，这一块，当处理的基数比较大的时候，推荐使用一些非高精度去重算法，Bloom过滤器，Hyper LogLog等。

汇总层也需要在每一条数据上增加一些额外字段到数仓里，这块与明细层一致，就不在单独讲解了。

数据质量保证

对于实时数仓数据质量的管理，我们通常由三步操作组成

第一步，数据与离线数据进行对比

首先，将汇总层数据Topic通过平台接入任务接入到离线仓库，然后通过数据质量任务，定时对实时数仓和离线数仓数据进行对比，并配置报警，数据差异，数据波动等。

第二步，配置报警，我们会在明细层以及汇总层，Topic配置生产监控，与以往数据波动，上游数据延迟或者积压，都需要进行报警。

第三部，构建实时血缘， Flink 在读取数据时候，会把信息读到flink catalog 这样就知道这个任务读取了哪个表，在解析客户DDL代码时，可以获得目标表信息，同步到我们的元数据服务。

参考文献：

美团实时数仓搭建：https://tech.meituan.com/2018/10/18/meishi-data-flink.html

菜鸟实时数仓：https://mp.weixin.qq.com/s/9ZRG76-vCM7AlRZNFCLrqA

本文目录

一. Sqoop介绍
二. Mysql 数据导入到 Hive
三. Hive数据导入到Mysql
四. mysql数据增量导入hive

• 1 基于递增列Append导入
  • 1). 创建hive表
  • 2). 创建job
  • 3) 执行job
• 2 Lastmodified 导入实战
  • 1). 新建一张表
  • 2). 初始化hive表:

一. Sqoop介绍

Sqoop是一个用来将Hadoop和关系型数据库中的数据相互转移的工具，可以将一个关系型数据库（例如：MySQL、Oracle、Postgres等）中的数据导进到Hadoop的HDFS中，也可以将HDFS的数据导进到关系型数据库中。对于某些NoSQL数据库它也提供了连接器。Sqoop，类似于其他ETL工具，使用元数据模型来判断数据类型并在数据从数据源转移到Hadoop时确保类型安全的数据处理。Sqoop专为大数据批量传输设计，能够分割数据集并创建Hadoop任务来处理每个区块。

本文版本说明

hadoop版本 : hadoop-2.7.2
hive版本 : hive-2.1.0
sqoop版本:sqoop-1.4.6

二. Mysql 数据导入到 Hive

1). 将mysql的people_access_log表导入到hive表web.people_access_log，并且hive中的表不存在。
mysql中表people_access_log数据为：

1,15110101010,1577003281739,'112.168.1.2','https://www.baidu.com'
2,15110101011,1577003281749,'112.16.1.23','https://www.baidu.com'
3,15110101012,1577003281759,'193.168.1.2','https://www.taobao.com'
4,15110101013,1577003281769,'112.18.1.2','https://www.baidu.com'
5,15110101014,1577003281779,'112.168.10.2','https://www.baidu.com'
6,15110101015,1577003281789,'11.168.1.2','https://www.taobao.com'

将mysql数据导入hive的命令为：

sqoop import \
--connect jdbc:mysql://master1.hadoop:3306/test \
--username root \
--password 123456 \
--table people_access_log \
-m 1 \
--hive-import \
--create-hive-table \
--fields-terminated-by '\t' \
--hive-table web.people_access_log

该命令会启用一个mapreduce任务，将mysql数据导入到hive表，并且指定了hive表的分隔符为\t，如果不指定则为默认分隔符^A(ctrl+A)。

参数说明

2). 也可以通过--query条件查询Mysql数据，将查询结果导入到Hive

sqoop import \
--connect jdbc:mysql://master1.hadoop:3306/test \
--username root \
--password 123456 \
--query 'select * from people_access_log where \$CONDITIONS and url = "https://www.baidu.com"' \
--target-dir /user/hive/warehouse/web/people_access_log \
--delete-target-dir \
--fields-terminated-by '\t' \
-m 1

三. Hive数据导入到Mysql

还是使用上面的hive表web.people_access_log,将其导入到mysql中的people_access_log_out表中.

sqoop export \
--connect jdbc:mysql://master1.hadoop:3306/test \
--username root \
--password 123456 \
--table people_access_log_out \
--input-fields-terminated-by '\t' \
--export-dir /user/hive/warehouse/web.db/people_access_log \
--num-mappers 1

注意：mysql表people_access_log_out需要提前建好，否则报错：ErrorException: Table 'test.people_access_log_out' doesn't exist。如果有id自增列，hive表也需要有，hive表与mysql表字段必须完全相同。

create table people_access_log_out like people_access_log;

执行完一个mr任务后，成功导入到mysql表people_access_log_out中.

四. mysql数据增量导入hive

实际中mysql数据会不断增加，这时候需要用sqoop将数据增量导入hive，然后进行海量数据分析统计。增量数据导入分两种，一是基于递增列的增量数据导入（Append方式）。二是基于时间列的增量数据导入（LastModified方式）。有几个核心参数：

• –check-column:用来指定一些列，这些列在增量导入时用来检查这些数据是否作为增量数据进行导入，和关系型数据库中的自增字段及时间戳类似.注意:这些被指定的列的类型不能使任意字符类型，如char、varchar等类型都是不可以的，同时–check-column可以去指定多个列
• –incremental:用来指定增量导入的模式，两种模式分别为Append和Lastmodified
• –last-value:指定上一次导入中检查列指定字段最大值

1. 基于递增列Append导入

接着前面的日志表，里面每行有一个唯一标识自增列ID，在关系型数据库中以主键形式存在。之前已经将id在0~6之间的编号的订单导入到Hadoop中了（这里为HDFS），现在一段时间后我们需要将近期产生的新的订 单数据导入Hadoop中（这里为HDFS），以供后续数仓进行分析。此时我们只需要指定–incremental 参数为append，–last-value参数为6即可。表示只从id大于6后即7开始导入。

1). 创建hive表

首先我们需要创建一张与mysql结构相同的hive表，假设指定字段分隔符为\t，后面导入数据时候分隔符也需要保持一致。

2). 创建job

增量导入肯定是多次进行的，可能每隔一个小时、一天等，所以需要创建计划任务，然后定时执行即可。我们都知道hive的数据是存在hdfs上面的，我们创建sqoop job的时候需要指定hive的数据表对应的hdfs目录，然后定时执行这个job即可。

当前mysql中数据，hive中数据与mysql一样也有6条：

增量导入有几个参数，保证下次同步的时候可以接着上次继续同步.

sqoop job --create mysql2hive_job -- import \
--connect jdbc:mysql://master1.hadoop:3306/test \
--username root \
--password 123456 \
--table people_access_log \
--target-dir /user/hive/warehouse/web.db/people_access_log \
--check-column id \
--incremental append \
--fields-terminated-by '\t' \
--last-value 6 \
-m 1

这里通过sqoop job --create job_name命令创建了一个名为mysql2hive_job的sqoop job。

3). 执行job

创建好了job，后面只需要定时周期执行这个提前定义好的job即可。我们先往mysql里面插入2条数据。

INSERT INTO `people_access_log` (`id`,`user_id`,`access_time`,`ip`,`url`) VALUES
(7,15110101016,1577003281790,'112.168.1.3','https://www.qq.com'),
(8,15110101017,1577003281791,'112.1.1.3','https://www.microsoft.com');

这样mysql里面就会多了2条数据。此时hive里面只有id为1 ~ 6的数据，执行同步job使用以下命令。

sqoop job -exec mysql2hive_job

执行完成后，发现刚才mysql新加入的id为7 ~ 8的两条数据已经同步到hive。

hive> select * from web.people_access_log;
OK
1   15110101010 1577003281739   112.168.1.2 https://www.baidu.com
2   15110101011 1577003281749   112.16.1.23 https://www.baidu.com
3   15110101012 1577003281759   193.168.1.2 https://www.taobao.com
4   15110101013 1577003281769   112.18.1.2  https://www.baidu.com
5   15110101014 1577003281779   112.168.10.2    https://www.baidu.com
6   15110101015 1577003281789   11.168.1.2  https://www.taobao.com
7   15110101016 1577003281790   112.168.1.3 https://www.qq.com
8   15110101017 1577003281791   112.1.1.3   https://www.microsoft.com

由于实际场景中，mysql表中的数据，比如订单表等，通常是一致有数据进入的，这时候只需要将sqoop job -exec mysql2hive_job这个命令定时(比如说10分钟频率)执行一次，就能将数据10分钟同步一次到hive数据仓库。

2. Lastmodified 导入实战

append适合业务系统库，一般业务系统表会通过自增ID作为主键标识唯一性。Lastmodified适合ETL的数据根据时间戳字段导入，表示只导入比这个时间戳大，即比这个时间晚的数据。

1). 新建一张表

在mysql中新建一张表people_access_log2，并且初始化几条数据:

CREATE TABLE `people_access_log2` (
  `id` bigint(20) unsigned NOT NULL AUTO_INCREMENT COMMENT 'id',
  `user_id` bigint(20) unsigned NOT NULL COMMENT '用户id',
  `access_time` timestamp DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
  `ip` varchar(15) NOT NULL COMMENT '访客ip',
  `url` varchar(200) CHARACTER SET utf8mb4 COLLATE utf8mb4_bin NOT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4;

插入数据:

insert into people_access_log2(id,user_id, ip, url) values(1,15110101010,'112.168.1.200','https://www.baidu.com');
insert into people_access_log2(id,user_id, ip, url) values(2,15110101011,'112.16.1.2','https://www.baidu.com');
insert into people_access_log2(id,user_id, ip, url) values(3,15110101012,'112.168.1.2','https://www.taobao.com');
insert into people_access_log2(id,user_id, ip, url) values(4,15110101013,'112.168.10.2','https://www.baidu.com');
insert into people_access_log2(id,user_id, ip, url) values(5,15110101014,'112.168.1.2','https://www.jd.com');
insert into people_access_log2(id,user_id, ip, url) values(6,15110101015,'112.168.12.4','https://www.qq.com');

mysql里面的数据就是这样：

2). 初始化hive表:

初始化hive数据，将mysql里面的6条数据导入hive中，并且可以自动帮助我们创建对应hive表，何乐而不为，否则我们需要自己手动创建，完成初始化工作。

sqoop import \
--connect jdbc:mysql://master1.hadoop:3306/test \
--username root \
--password 123456 \
--table people_access_log2 \
--hive-import \
--create-hive-table \
--fields-terminated-by ',' \
--hive-table web.people_access_log2

可以看到执行该命令后，启动了二一个mapreduce任务，这样6条数据就进入hive表web.people_access_log2了:

hive> select * from web.people_access_log2;
OK
1   15110101010 2019-12-28 16:23:10.0   112.168.1.200   https://www.baidu.com
2   15110101011 2019-12-28 16:23:33.0   112.16.1.2  https://www.baidu.com
3   15110101012 2019-12-28 16:23:41.0   112.168.1.2 https://www.taobao.com
4   15110101013 2019-12-28 16:23:46.0   112.168.10.2    https://www.baidu.com
5   15110101014 2019-12-28 16:23:52.0   112.168.1.2 https://www.jd.com
6   15110101015 2019-12-28 16:23:56.0   112.168.12.4    https://www.qq.com
Time taken: 0.326 seconds, Fetched: 6 row(s)

3). 增量导入数据:

我们再次插入一条数据进入mysql的people_access_log2表:

insert into people_access_log2(id,user_id, ip, url) values(7,15110101016,'112.168.12.45','https://www.qq.com');

此时，mysql表里面已经有7条数据了，我们使用incremental的方式进行增量的导入到hive:

sqoop import \
--connect jdbc:mysql://master1.hadoop:3306/test \
--username root \
--password 123456 \
--table people_access_log2 \
--hive-import \
--hive-table people_access_log2 \
-m 1 \
--check-column access_time \
--incremental lastmodified \
--last-value "2019-12-28 16:23:56" \

2019-12-28 16:23:56就是第6条数据的时间，这里需要指定。报错了:

19/12/28 16:17:25 ERROR tool.ImportTool: Error during import: --merge-key or --append is required when using --incremental lastmodified and the output directory exists.

注意：可以看到--merge-key or --append is required when using --incremental lastmodified意思是，这种基于时间导入模式，需要指定--merge-key或者--append参数，表示根据时间戳导入，数据是直接在末尾追加(append)还是合并(merge)，这里使用merge方式，根据id合并：

sqoop import \
--connect jdbc:mysql://master1.hadoop:3306/test \
--username root \
--password 123456 \
--table people_access_log2 \
--hive-import \
--hive-table web.people_access_log2 \
--check-column access_time \
--incremental lastmodified \
--last-value "2019-12-28 16:23:56" \
--fields-terminated-by ',' \
--merge-key id

执行该命令后，与直接导入不同，该命令启动了2个mapreduce任务，这样就把数据增量merge导入hive表了.

hive> select * from web.people_access_log2 order by id;
OK
1   15110101010 2019-12-28 16:23:10.0   112.168.1.200   https://www.baidu.com
2   15110101011 2019-12-28 16:23:33.0   112.16.1.2  https://www.baidu.com
3   15110101012 2019-12-28 16:23:41.0   112.168.1.2 https://www.taobao.com
4   15110101013 2019-12-28 16:23:46.0   112.168.10.2    https://www.baidu.com
5   15110101014 2019-12-28 16:23:52.0   112.168.1.2 https://www.jd.com
6   15110101015 2019-12-28 16:23:56.0   112.168.12.4    https://www.qq.com
6   15110101015 2019-12-28 16:23:56.0   112.168.12.4    https://www.qq.com
7   15110101016 2019-12-28 16:28:24.0   112.168.12.45   https://www.qq.com
Time taken: 0.241 seconds, Fetched: 8 row(s)

可以看到id=6的数据，有2条，它的时间刚好是--last-value指定的时间，则会导入大于等于--last-value指定时间的数据，这点需要注意。

转载请注明：柯广的网络日志 » sqoop用法之mysql与hive数据导入导出