汇总:请教网奇系统新闻采集规则。

　　汇总:请教网奇系统新闻采集规则。

　　这应该是你的采集系统没有工作

　　

" />

　　顺便再说一句，你是自用还是公用？如果是后者，我建议你立即切换到官方新闻采集系统。

　　因为网上找的免费采集软件基本都不好用，我也在用采集软件，那些所谓的免费软件都是个人开发的小玩意儿，我只能玩玩。你可以制作一个官方软件的试用版，比那些免费的好太多了！

　　

" />

　　网络信息采集是信息化进程中的重要一步，所以很多公司都在推广**优财云采集

、**collector等软件，但这些都是应用层次比较浅的技术。很多地方是做不到的，比如网站限制你采集

等问题。真正想要高端的技术或者成熟的产品，一定是那些在采集

行业非常专业的公司。深圳的乐思软件在国内网络信息采集方面比较专业，大家可以去他们的网站了解一下。

　　汇总:批流一体数据集成工具ChunJun同步Hive事务表原理详解及实战分享

　　本期我们就带大家回顾一下《春君同步Hive交易表详解》的直播分享

　　Hive事务表结构及原理

　　Hive 是一个基于 Hadoop 的数据仓库工具，用于数据提取、转换和加载。它是一种可以存储、查询和分析存储在 Hadoop 中的*敏*感*词*数据的机制。Hive数据仓库工具可以将结构化数据文件映射成数据库表，并提供SQL查询功能，可以将SQL语句转换为MapReduce任务执行。

　　在分享Hive事务表的具体内容之前，我们先了解一下Hive事务表在HDFS存储上的一些局限性。

　　虽然 Hive 支持 ACID 语义的事务，但不如在 MySQL 中使用方便，并且有很多局限性。具体限制如下：

　　下面的矩阵包括可以使用 Hive 创建的表的类型、是否支持 ACID 属性、所需的存储格式和关键的 SQL 操作。

　　了解了Hive事务表的局限性之后，现在我们来详细了解一下Hive事务表的内容。

　　01

　　事务表文件名详解

　　基本目录：

　　$partition/base_$wid/$桶

　　增量目录：

　　$分区/delta_$wid_$wid_$stid/$桶

　　参数目录：

　　$partition/delete_delta_$wid_$wid_$stid/$bucket

　　02

　　交易表文件内容详解

　　$ orc-tools 数据 bucket_00000

　　{"operation":0,"originalTransaction":1,"bucket":536870912,"rowId":0,"currentTransaction":1,"row":{"id":1,"name":"Jerry", “年龄”：18}}

　　{"operation":0,"originalTransaction":1,"bucket":536870912,"rowId":1,"currentTransaction":1,"row":{"id":2,"name":"Tom", “年龄”：19}}

　　{"operation":0,"originalTransaction":1,"bucket":536870912,"rowId":2,"currentTransaction":1,"row":{"id":3,"name":"Kate", “年龄”：20}}

　　1.operation 0表示插入，1表示更新，2表示删除。由于使用了split-update，所以不会出现UPDATE。

　　2.originalTransaction是这条记录的原创

写事务ID：

　　一个。对于INSERT操作，这个值与currentTransaction一致；

　　b. 对于DELETE，是第一次插入记录时的写事务ID。

　　3. bucket是一个32位的整数，用BucketCodec编码，每个二进制位的含义是：

　　一个。1-3位：编码版本，目前为001；

　　b. 4位：预留；

　　C。Bits 5-16：Bucket ID，从0开始。Bucket ID由CLUSTERED BY子句中指定的字段和Bucket的个数决定。该值与bucket_N中的N一致；

　　d. 17-20位：预留；

　　e. 第 21-32 位：语句 ID；

　　例如整数536936448的二进制格式为0000000000，即编码为version 1的格式，bucket ID为1。

　　4.rowId是一个自增的唯一ID，在写事务和bucketing的结合中是唯一的；

　　5.currentTransaction 当前写事务ID；

　　6. 行特定数据。对于 DELETE 语句，它为空。

　　03

　　更新 Hive 事务表数据

　　UPDATE employee SET age = 21 WHERE id = 2;

　　该语句首先会查询所有符合条件的记录，获取其row_id信息，然后分别创建delete和delta目录：

　　/用户/配置单元/仓库/员工/delta_0000001_0000001_0000/bucket_00000

　　/user/hive/warehouse/employee/delete_delta_0000002_0000002_0000/bucket_00000（更新）

　　/user/hive/warehouse/employee/delta_0000002_0000002_0000/bucket_00000（更新）

　　delete_delta_0000002_0000002_0000/bucket_00000

　　收录

已删除的记录：

　　{“operation”：2，“originalTransaction”：1，“bucket”：536870912，“rowId”：1，“currentTransaction”：2，“row”：null}

　　delta_0000002_0000002_0000/bucket_00000

　　收录

更新的数据：

　　{"operation":0,"originalTransaction":2,"bucket":536870912,"rowId":0,"currentTransaction":2,"row":{"id":2,"name":"Tom", “工资”：21}}

　　04

　　如何查看Row_ID信息？

　　05

　　事务表压缩（Compact）

　　随着写操作的积累，表中的delta和delete文件会越来越多。在事务表的读取过程中需要合并所有文件。如果数量太多，效率会受到影响。另外，小文件对HDFS等文件系统也有危害。不够友好，所以Hive引入了compaction的概念，分为Minor和Major两大类。

　　●*敏*感*词*人

　　Minor Compaction会将所有delta文件压缩为一个文件，删除文件也会压缩为一个文件。生成的压缩文件名将包括写入事务 ID 范围并省略语句 ID。

　　压缩过程在 Hive Metastore 中运行，并根据特定阈值自动触发。我们也可以使用如下语句来手动触发：

　　更改表 dtstack COMPACT '次要'。

　　●专业

　　

" />

　　Major Compaction 会将所有增量文件和删除文件压缩到一个基本文件中。生成的压缩文件名将收录

所有写入事务 ID 中最大的事务 ID。

　　压缩过程在 Hive Metastore 中运行，并根据特定阈值自动触发。我们也可以使用如下语句来手动触发：

　　ALTER TABLE dtstack COMPACT '主要'。

　　06

　　文件内容详解

　　ALTER TABLE employee COMPACT 'minor';

　　语句执行前：

　　/用户/配置单元/仓库/员工/delta_0000001_0000001_0000

　　/user/hive/warehouse/employee/delta_0000002_0000002_0000（插入创建，mary的数据）

　　/user/hive/warehouse/employee/delete_delta_0000002_0000002_0001（更新）

　　/user/hive/warehouse/employee/delta_0000002_0000002_0001（更新）

　　语句执行后：

　　/用户/配置单元/仓库/员工/delete_delta_0000001_0000002

　　/用户/配置单元/仓库/员工/delta_0000001_0000002

　　07

　　读取 Hive 事务表

　　我们可以看到ACID事务表会收录

三种类型的文件，分别是base、delta和delete。文件中的每一行数据将由row__id 标识和排序。从 ACID 事务表中读取数据就是合并这些文件以获得最新事务的结果。这个过程在OrcInputFormat和OrcRawRecordMerger类中实现，本质上是一种归并排序算法。

　　以下面的文件为例，生成这些文件的操作是：

　　1.插入三条记录

　　2. 进行大压实

　　3.然后更新两条记录。

　　1-0-0-1 用于 originalTransaction - bucketId - rowId - currentTra

　　08

　　合并算法

　　将所有数据行按照(originalTransaction, bucketId, rowId)正序排列，(currentTransaction)倒序排列，即：

　　originalTransaction-bucketId-rowId-currentTransaction

　　(base_1)1-0-0-1

　　(delete_2)1-0-1-2# 被跳过（删除）

　　(base_1)1-0-1-1 # 跳过（当前记录的row_id(1)与之前的数据相同）

　　(delete_2)1-0-2-2 # 跳过 (DELETE)

　　(base_1)1-0-2-1 # 跳过（当前记录的row_id(2)与之前的数据相同）

　　(delta_2)2-0-0-2

　　(delta_2)2-0-1-2

　　获取第一条记录；

　　1、如果当前记录的row_id与上一条数据相同，则跳过；

　　2、如果当前记录的操作类型是DELETE，则跳过；

　　通过以上两条规则，对于1-0-1-2和1-0-1-1，这条记录将被跳过；

　　如果没有跳过，记录会向下游输出；

　　重复上述过程。

　　合并过程是流式的，即Hive会打开所有文件，预读第一条记录，并将row__id信息存入ReaderKey类型。

　　ChunJun读写Hive事务表实战

　　了解了Hive事务表的基本原理后，下面分享一下如何在ChunJun中读写Hive事务表。

　　01

　　交易表数据准备

　　-- 创建事务表

　　创建表 dtstack(

　　*敏*感*词*，

　　名称字符串，

　　年龄整数

　　)

　　存储为兽人

　　TBLPROPERTIES('交易'='真');

　　-- 插入10条测试数据

　　插入 dtstack (id, name, age)

　　值 (1, "aa", 11), (2, "bb", 12), (3, "cc", 13), (4, "dd", 14), (5, "ee", 15) ,

　　(6, "ff", 16), (7, "gg", 17), (8, "hh", 18), (9, "ii", 19), (10, "jj", 20);

　　02

　　配置 ChunJun json 脚本

　　

" />

　　03

　　提交任务（读写事务表）

　　# 开始会话

　　/root/wujuan/flink-1.12.7/bin/yarn-session.sh -t $ChunJun_HOME -d

　　#提交 Yarn Session 任务

　　# 读取交易表

　　/root/wujuan/ChunJun/bin/ChunJun-yarn-session.sh -job /root/wujuan/ChunJun/ChunJun-examples/json/hive3/hive3_transaction_stream.json -confProp {\"yarn.application.id\":\ “application_32_0134\”}

　　#写交易表

　　/root/wujuan/ChunJun/bin/ChunJun-yarn-session.sh -job /root/wujuan/ChunJun/ChunJun-examples/json/hive3/stream_hive3_transaction.json -confProp {\"yarn.application.id\":\ “application_32_0134\”}

　　根据上一行的结果替换yarn.application.id

　　ChunJun读写Hive事务表源码分析

　　压缩器是一组在 Metastore 内部运行以支持 ACID 系统的守护进程。它由 Initiator、Worker、Cleaner、AcidHouseKeeperService 等组成。

　　01

　　压路机

　　●增量文件压缩

　　在表的不断修改中，会创建越来越多的delta文件，需要对这些文件进行压缩以保证性能。有两种类型的压缩，(minor) small compression 和 (major) major compression：

　　所有压缩工作都在后台执行，不会阻止并发读取和写入数据。压缩后，系统等待所有旧文件被读取，然后删除旧文件。

　　●发起人

　　该模块负责发现要压缩的表或分区。这应该在 Metastore 中使用 pactor.initiator.on 启用。每个 Compact 任务处理一个分区（如果表未分区，则处理整个表）。如果一个分区连续compaction失败的次数超过pacts.threshold，这个分区的自动compaction调度就会停止。

　　●工人

　　每个 Worker 处理一个压缩任务。压缩是具有以下形式名称的 MapReduce 作业。-压实机-..

　　. 每个 Worker 将作业提交到集群（如果定义了 pactor.job.queue），并等待作业完成。pactor.worker.threads 确定每个 Metastore 中的工作人员数量。Hive 仓库中的 worker 总数决定了并发压缩的最大数量。

　　●清洁工

　　这个过程是在确定压缩后不再需要之后删除delta文件。

　　●酸管家服务

　　此过程查找尚未在 hive.txn.timeout 内被心跳的事务并中止它们。系统假设发起事务的客户端在停止心跳后崩溃了，它锁定的资源应该被释放。

　　●显示压实

　　此命令显示有关当前正在运行的压缩和最近的压缩历史记录的信息（具有可配置的保留期）。此历史记录表明它自 HIVE-12353 以来一直可用。

　　●紧凑的按键配置

　　02

　　如何调试蜂巢

　　1.调试蜂巢客户端

　　配置单元--调试

　　2.调试hive metastore

　　hive --service metastore --debug:port=8881,mainSuspend=y,childSuspend=n --hiveconf hive.root.logger=DEBUG,控制台

　　3.调试hive mr任务

　　03

　　读写过滤和CompactorMR排序的关键代码

　　04

　　Minor&Major组合源码（CompactorMR Map类）

　　纯君文件系统未来规划

　　最后介绍一下春君文件系统未来的规划：

　　●基于FLIP-27优化的文件系统

　　批量流统一实现，线程模型简单，分片和读数据分离。

　　● Hive分片优化

　　分片更细粒度更细，充分发挥并发能力

　　●改进Exactly Once的语义

　　增强的异常鲁棒性。

　　●HDFS文件系统断点续传

　　根据分区、文件数、文件行数等确定端点的位置，并将状态存储在检查点中。

　　●实时采集文件

　　实时监控目录中的多个附加文件。

　　●文件系统格式的通用性

　　JSON、CSV、Text、XM、EXCELL公共包统一提取。