背景

最近因公司的CRM项目用的mongodb,浏览了所有旧代码，看到特别多多表原子操作的问题。借此机会就来看看mongodb4.0后出来的副本集事务的能力。

问题还原

多表操作场景

涉及异常回滚的原子性问题,

// 创建流水号（流水号表自增）

throw Error -> 异常中断 

// 创建订单

问题：当异常中断时候，实际流水号表已经成功自增1，但是创建订单失败。当下一次进行操作的时候，实际流水号缺失了1位的订单流水号。

这里其实还隐藏着一个问题，高并发未加锁，会导致流水号异常。

这就是事务的原子性，实际应该当异常中断，启动事务回滚，回滚流水号的自增创建。

问题还原2

redis版本异常


// 订单入库
await ctx.model.Delivery.VirtualOrder.create(orders)

// 发送通知  -> 异常出现redis报错
const jobs = await ctx.app.noticeQueue.addBulk(datas)

问题：当异常redis中断导致发送消息失败，应该启动事务回滚

正确操作：启动事务，进行异常回滚，如下

app/extend/context.js

module.exports = {
  async getSession(
    opt = {
      readPreference: { mode: 'primary' },
    }
  ) {
    const { mongoose } = this.app
    const session = await mongoose.startSession(opt)
    await session.startTransaction({
      readConcern: { level: 'majority' },
      writeConcern: { w: 'majority' },
    })
    return session
  },
}


const { ctx } = this
const session = await this.ctx.getSession()
try {
  // 订单入库
  await ctx.model.Delivery.VirtualOrder.create(orders, { session })
  // 发送通知  -> 异常出现redis报错
  const jobs = await ctx.app.noticeQueue.addBulk(datas)
 } catch (error) {
  console.log('开始回滚')
  await session.abortTransaction()
  session.endSession()
  console.log(error)
}

错误或不建议的操作：人工删除

try {
  // 创建
  // 创建
  // 创建
  // 创建
  // 创建
  // 创建
} catch{
  // 人工删除创建
}

mongodb副本集环境搭建

bash 脚本

#!/bin/bash
# 生成 keyfile
mkdir ./keyfile
openssl rand -base64 745 > ./keyfile/mongoReplSet-keyfile
chmod 600 ./keyfile/mongoReplSet-keyfile

出现如下错误：

configsvr01    | {"t":{"$date":"2021-05-29T17:38:02.750+00:00"},"s":"I",  "c":"ACCESS",   "id":20254,   "ctx":"main","msg":"Read security file failed","attr":{"error":{"code":30,"codeName":"InvalidPath","errmsg":"error opening file: /data/mongo.key: bad file"}}}

解决办法：变更mongoReplSet-keyfile 所属用户
chown 999 mongoReplSet-keyfile

启动 Docker

docker-compose -f docker-compose.yml up -d
docker-compose

version: '3.1'
services:
  mongo1:
    image: mongo
    hostname: mongo1
    container_name: mongo1
    restart: always
    ports:
      - 27011:27017
    volumes:
      - ./keyfile:/data/keyfile
    environment:
      MONGO_INITDB_ROOT_USERNAME: root
      MONGO_INITDB_ROOT_PASSWORD: 123123
    command: mongod --auth --keyFile /data/keyfile/mongoReplSet-keyfile --bind_ip_all --replSet rs0

  mongo2:
    image: mongo
    hostname: mongo2
    container_name: mongo2
    restart: always
    ports:
      - 27012:27017
    volumes:
      - ./keyfile:/data/keyfile
    environment:
      MONGO_INITDB_ROOT_USERNAME: root
      MONGO_INITDB_ROOT_PASSWORD: 123123
    command: mongod --auth --keyFile /data/keyfile/mongoReplSet-keyfile --bind_ip_all --replSet rs0

  mongo3:
    image: mongo
    hostname: mongo3
    container_name: mongo3
    restart: always
    ports:
      - 27013:27017
    volumes:
      - ./keyfile:/data/keyfile
    environment:
      MONGO_INITDB_ROOT_USERNAME: root
      MONGO_INITDB_ROOT_PASSWORD: 123123
    command: mongod --auth --keyFile /data/keyfile/mongoReplSet-keyfile --bind_ip_all --replSet rs0

容器名	ip	备注
mongo1	10.8.99.44:27011	Primary(主, 读写)
mongo2	10.8.99.44:27012	Secondary1(从,读)
mongo3	10.8.99.44:27013	Secondary2(从, 读)

配置副本集

docker exec -it <container> mongo
mongo -u root -p 123123
use admin
rs.initiate(
  {
    _id : 'rs0',
    members: [
      { _id : 0, host : "10.8.99.44:27011", priority:3 },
      { _id : 1, host : "10.8.99.44:27012", priority:3 },
      { _id : 2, host : "10.8.99.44:27013", priority:3 }
    ]
  }
)

事务实践

如何通过mongoose连接副本集

config.mongoose = {
  url: 'mongodb://10.8.99.44:27011,10.8.99.44:27012,10.8.99.44:27013/log',
  options: {
    auth: {
      user: 'admin',
      password: '123123',
    },
    replicaSet: 'rs0', // 副本集
    readPreference: 'secondaryPreferred', // 读哪个副本集 从哪个副本集读数据
    readConcern: { level: 'majority' }, // 读的策略 解决脏读 决定这个节点上的数据哪些是可读的，类似于关系数据库的隔离级别 读取在大多数节点上提交完成的数据;
    // write concern 当配置majority时，大多数节点为2，Primary感知到两个节点被写入完成时就代表成功写入
    writeConcern: {
      w: 'majority',
      // j，该参数表示是否写操作要进行journal持久化之后才向用户确认；
      // {j: true} 要求primary写操作进行了journal持久化之后才向用户确认；
      // {j: false} 要求写操作已经在journal缓存中即可向用户确认；journal后续会将持久化到磁盘，默认是100ms；
      // j: true,
      j: true,
      wtimeout: 1000,
    },
    keepAlive: true,
    keepAliveInitialDelay: 300000,
    useNewUrlParser: true,
    useFindAndModify: false,
    useCreateIndex: true,
    useUnifiedTopology: true,
    serverSelectionTimeoutMS: 30000,
    socketTimeoutMS: 45000,
  },
}

readPreference

默认情况下，读写都指定到副本集中的 Primary 节点。对于读多写少的情况我们可以使用读写分离来减轻 DB 的压力。MongoDB 驱动程序支持五种读取首选项(Read Preference) 模式。

Read Preference	描述
primary	默认模式。所有操作都从当前副本集 primary 读取。
primaryPreferred	在大多数情况下，从 primary 读取，但如果不可用，则从 secondary 读取。
secondary	所有操作都从 secondary 中读取。
secondaryPreferred	在大多数情况下，从 secondary 读取，但如果没有 secondary 可用，则从 primary 读取。
nearest	无论成员的类型如何，操作都从具有最小网络延迟的副本集成员读取

mongoose事务回滚

app/extend/context.js

module.exports = {
  async getSession(
    opt = {
      readPreference: { mode: 'primary' },
    }
  ) {
    const { mongoose } = this.app
    const session = await mongoose.startSession(opt)
    await session.startTransaction({
      readConcern: { level: 'majority' },
      writeConcern: { w: 'majority' },
    })
    return session
  },
}

app/service/test.js

async transaction() {
    const { ctx } = this
    const session = await this.ctx.getSession()
    const db = this.app.mongoose.connection
    try {
      const res1 = await ctx.model.VpnLog.create(
        [
          {
            mac_address: 'test',
            session: 'test',
            source_ip: 'test',
            destination_ip: 'test',
            hub: 'test',
            items_per_second: 111,
            created_at: new Date(),
          },
        ],
        { session }
      )
      throw new Error('出错')   -> 出错，如其他程序的操作错误
      const res2 = await ctx.model.VpnLog.create(
        [
          {
            mac_address: 'test2',
            session: 'test2',
            source_ip: 'test2',
            destination_ip: 'test2',
            hub: 'test2',
            items_per_second: 222,
            created_at: new Date(),
          },
        ],
        { session }
      )
      await session.commitTransaction()
      session.endSession()
      return {
        res1,
        res2,
      }
    } catch (error) {
      console.log('开始回滚')
      await session.abortTransaction()
      session.endSession()
      console.log(error)
    }
  }

可以看看res1 是不是回滚创建，在数据库中找不到了

前言

众所周知，mongoose的日期格式是ISODate,也就是使用的utc时间,举个栗子：2020-12-11T16:00:00.000Z ,T表示分隔符，Z表示的是UTC。

UTC：世界标准时间，在标准时间上加上8小时，即东八区时间，也就是北京时间。

咱举个例子:

北京时间：2020-12-12 00:00:00对应的国际标准时间格式为：2020-12-11T16:00:00.000Z。

当我们的前端页面通过接口拿到我的utc时间后，一般通过new Date(时间)，就能快速的转换成当地的时间。

这些周知的我就不再多举例了。

笔记原因

做这个笔记前,我遇到了时间进入到数据库没有准确的转换为utc,于是，好奇心驱使，我们开启mongoose的debug模式，来看看是什么实际mongoose到原生层的实际过程。

1 2	// 开启mongoose调试 this.app.mongoose.set('debug', true)

举例，我有一个schema，里面有个updateDate,我们先来看看不同的日期插入数据库时候，实际的表现。

module.exports = (app) => {
  const mongoose = app.mongoose
  const Schema = mongoose.Schema

  const FileSchema = new Schema(
    {
      fileName: { type: String }, // 文件名
      uploadDate: { type: Date }, // 上传创建日期
      creatorName: { type: String }, // 创建者姓名
    },
    {
      collection: 'sys_files',
    }
  )

  return mongoose.model('File', FileSchema)
}

方式一：使用new Date 插入

const deliver = await ctx.model.Delivery.File.create({
  fileName: '文件名',
  uploadDate: new Date("2021-12-12 00:00:00"),
  creatorName: "张三",
})

3suLxw

通过debug日志：

Mongoose: sys_files.insertOne({ isDel: false, _id: ObjectId("61b5599cba1fcfeeb79c57cd"), fileName: '文件名', uploadDate: new Date("Sat, 11 Dec 2021 16:00:00 GMT"), creatorName: '张三', __v: 0}, { session: null })

结论：`

我们发现mongoose的ORM层的create实际调用了insertOne,插入的本地时间new Date("2021-12-12 00:00:00")（2021-12-11T16:00:00.000Z）到达原生层变成了Sat, 11 Dec 2021 16:00:00 GMT并在进行了一次new Date(),

所以整个orm层到mongodb原生层的过程是这样的：
日期传入-> (ORM)转换为GMT零时区-> (ORM)new Date()转为ISODate -> 入库

->new Date("2021-12-12 00:00:00")传入 -> 2021-12-11T16:00:00.000Z
->(ORM)转换为GMT零时区 Sat, 11 Dec 2021 16:00:00 GMT
->(ORM)new Date()转为ISODate new Date("Sat, 11 Dec 2021 16:00:00 GMT") -> 2020-12-11T16:00:00.000Z
-> 入库

我们在拿一个字符串时间和moment时间对象来校验是不是这个过程：日期传入-> (ORM)转换为GMT零时区-> (ORM)new Date()转为ISODate -> 入库

方式二：使用String 字符串插入


const deliver = await ctx.model.Delivery.File.create({
  fileName: '文件名',
  uploadDate: "2021-12-12 00:00:00",
  creatorName: "李四",
})

FXDRVZ

结论：

->2021-12-12 00:00:00传入
->(ORM)转换为GMT零时区 Sat, 11 Dec 2021 16:00:00 GMT
->(ORM)new Date()转为ISODatenew Date("Sat, 11 Dec 2021 16:00:00 GMT") -> 2020-12-11T16:00:00.000Z
-> 入库

方式三：使用moment 插入


const deliver = await ctx.model.Delivery.File.create({
  fileName: '文件名',
  uploadDate: moment("2021-12-12 00:00:00"),
  creatorName: '王五',
})

k6gMeQ

结论：

->moment("2021-12-12 00:00:00")传入 -> Moment<2021-12-12T00:00:00+08:00>
->(ORM)转换为GMT零时区 Sat, 11 Dec 2021 16:00:00 GMT
->(ORM)new Date()转为ISODatenew Date("Sat, 11 Dec 2021 16:00:00 GMT") -> 2020-12-11T16:00:00.000Z
-> 入库

others

我们发现当我们的时间只要是精确到时分秒,进入到mongodb数据库后，都能正确的转换成UTC时间。

那我们来试试年月日的情况

const file = await ctx.model.Delivery.File.create({
  fileName: '文件名',
  uploadDate: new Date('2021-12-12'),
  creatorName: '刘九',
})

L8L1y4

结论：`

->new Date('2021-12-12')传入 -> 2021-12-12T00:00:00.000Z （注意此处年月日时间转换为UTC时间与上面带时分秒的差异）
->(ORM)转换为GMT零时区 Sat, 11 Dec 2021 00:00:00 GMT
->(ORM)new Date()转为ISODatenew Date("Sat, 11 Dec 2021 00:00:00 GMT") -> 2020-12-12T00:00:00.000Z
-> 入库

这也就解释了为什么本人在项目中传入2021-12-12 的日期最终却变成了utc 2020-12-12T00:00:00.000Z,也就是为什么new Date()本地时间会多出来8个小时的原因了。

总结

mongoose这个ORM实际做了一步强制new Date()转换为utc时间。所以无论传入什么本地时间，都会强制转换mongodb所需要的ISODate时期格式。
所以无论是moment、dayjs等时间库的时间，最后都会被momgoose强制转换为new Date 的UTC时间。与用什么时间库或时间格式并无直接关系。

番外篇

我们来验证下查询的时候，传入的时间是不是也会通过mongoose自动强制new Date
De6b3B
uuLCtR

验证发现流程：

-> 传入时间
->(ORM)转换为GMT零时区 Sat, 11 Dec 2021 00:00:00 GMT
->(ORM)new Date()转为ISODatenew Date("Sat, 11 Dec 2021 00:00:00 GMT") -> 2020-12-12T00:00:00.000Z
-> 查询

前言

我们在 mongoose关联查询技巧笔记 的关联技巧一：关联引用子查父 中我们发现可以通过lookup进行子查父,那么我们是否有什么快捷方式能在子当中定义一个字段，查询子的时候就能带出父的信息呢？

virtual使用

virtual虚拟值填充-官方文档

schema 设置 toJSON 和toObject可获取

toJSON: {
  virtuals: true,
},
toObject: {
  virtuals: true,
},

合并字段

personSchema.virtual('fullName').get(function () {
  return this.name.first + ' ' + this.name.last;
});

var Person = mongoose.model('Person', personSchema);

var p = new Person({
  name: { first: 'junyao', last: 'hong' }
});

p.name.fullName // junyao hong

virtual关联查询(子查父)

FcvdNf

BookSchema.virtual('author', {
  ref: 'authors',
  localField: '_id',
  foreignField: 'books',
  justOne: true,
})

前言

数据库设计中数据之间的关联关系是极其常见的：一对一、一对多、多对多，作为 NoSQL 领头羊的 MongoDB 中常用做法无非「内嵌」和「引用」两种，因为 Document 有 16MB 的大小限制且「内嵌」不适合复杂的多对多关系，「引用」是用得更广泛的关联方式，所以 MongoDB 官方称其为“Normalized Data Models”——标准化数据模型。

引用式的关联其实很简单，指文档与文档之间通过_id字段的引用来进行关联。Mongoose 4.5.0版本以后提供了与 aggregate 功能写法都非常类似的virtual()方法,这里先不做对比,可查看另一篇笔记mongoose4.5之virtual虚拟值填充，本文要阐述的重点就在于如何去查这两个表通过aggregate 与 populate，并介绍4.5的aggregate的关联引用反向查询。

populate

先说说populate吧，首先，Mongoose 的一切始于 Schema，使用 populate 的重点也在于 Schema 中的设置：

const mongoose = require('mongoose');
const Schema = mongoose.Schema;
const authorSchema = new Schema({
    "author": String,
    "age": Number,
    "books": [{
        type: Schema.Types.ObjectId,
        ref: 'Book' // 关联的Model
    }]
});
module.exports = mongoose.model("Author", authorSchema, "authors"); // 分别为Model名、Schema、数据库中集合名

使用

1
2
3

let result = await Author.find({
    "author": "Zander"
  }).populate("books");

返回如下数据

[
        {
            "books": [
                {
                    "_id": "5dccfcb5a3fab06c89020c8d",
                    "name": "代码的弱点"
                },
                {
                    "_id": "5dccfd30a3fab06c89020caa",
                    "name": "代码与六便士"
                },
                {
                    "_id": "5dccfda6a3fab06c89020cc3",
                    "name": "代码失格"
                }
            ],
            "_id": "5dccfc3aa3fab06c89020c65",
            "author": "Zander",
            "age": 18
        }
    ]

aggregate

FcvdNf

使用 aggregate 实现聚合查询作者 Zander 的基本信息及其所有著作信息：

let result = await Author.aggregate([{ // 操作的Model为Author
    $lookup: {
      from: "books", // 数据库中关联的集合名
      localField: "books", // author文档中关联的字段
      foreignField: "_id", // book文档中关联的字段
      as: "bookList" // 返回数据的字段名
    }
  }, {
    $match: { // 筛选条件
      "author": "Zander"
    }
  }]);

返回数据：

[
        {
            "_id": "5dccfc3aa3fab06c89020c65",
            "author": "Zander",
            "age": 18,
            "bookList": [
                {
                    "_id": "5dccfcb5a3fab06c89020c8d",
                    "name": "代码的弱点"
                },
                {
                    "_id": "5dccfd30a3fab06c89020caa",
                    "name": "代码与六便士"
                },
                {
                    "_id": "5dccfda6a3fab06c89020cc3",
                    "name": "代码失格"
                }
            ]
        }
    ]

对比

1. 灵活性

现在可以观察到的就是 aggregate 灵活的点在于可以更改关联查询后返回数据的 key（返回数据中的bookList），而 populate 返回数据的 key 只能是原来的字段名（返回数据中的books）。值得一提的是 aggregate 更擅长在聚合管道中对数据进行二次处理，比如$unwind拆分、$group分组等等。

2. 功能性

此外，还有一种情况：依旧是上面的数据，如果要根据著作 name 找到著作信息和作者信息，使用 aggregate 的$lookup只需要这样就做到了😏：

$lookup: {
  from: "authors",
  localField: "_id",
  foreignField: "books",
  as: "author"
}

然而 populate：“我太难了！” 是的，它做不到这种使用_id实现的反向关联查询.

3. 性能方面

看完了外表再说说内在——查询性能，populate 实际是DBRef[^4]的引用方式，相当于多构造了一层查询。比如有10条数据，在find()查询到了主集合内的10条数据后会再进行populate()引用的额外10条数据的查询，性能也相对的大打折扣了。这里有位大佬对aggregate()和find()进行了性能上的对比，结论也显而易见——比 find 查询速度都快的 aggregate 比关联查询的 find + populate 定是有过之而无不及了。

总结

	aggregation	populate
灵活性	⭐️⭐️⭐️⭐️⭐	⭐️
反向关联	⭐️⭐️⭐️⭐️⭐️	⭐️⭐️
功能性	⭐️⭐️⭐️⭐️⭐️	⭐️⭐️⭐️
代码简洁度	⭐️	⭐️⭐️⭐️⭐️⭐️
查询性能	⭐️⭐️⭐️⭐️	⭐️⭐️

综合来看，aggregate 在多集合关联查询和对查询数据的二次处理方面更优，而 populate 更适合简单的正向关联关系且其形成的代码样式较优雅，可读性高而易于维护，性能方面的考究对日常开发中的普通应用来说则大可忽略不计。

关联技巧一：关联引用子查父

FcvdNf

$lookup: {
  from: "authors",
  localField: "_id",
  foreignField: "books",
  as: "author"
}

也可通过 mongoose4.5之virtual虚拟值填充快速获取父内容

关联技巧二：关联数组转对象

const lookup = {
  $lookup: {
    from: 'sys_invoices',
    localField: 'invoice',
    foreignField: '_id',
    as: 'invoice',
  },
}

const project = {
  $project: {
    invoice: { $arrayElemAt: ['$invoice', 0] },
  },
}

elasticSearch-mapping相关

发表于 2021-06-08 更新于 2024-11-02 阅读次数：

字段的数据类型

简单类型
- Text / Keyword
- Date
- Integer / Floating
- Boolean
- IPv4 & IPv6
复杂类型 - 对象和嵌套对象
- 对象类型 / 嵌套类型
特殊类型
- geo_point & geo_shape / percolator

Dynamic Mapping

写入文档的时候，如果索引不存在，会自动创建索引
Dynamic Mapping 的机制，使得我们无需手动定义 Mappings。Elasticsearch 会自动根据文档信息，推算出字段的类型
但是会有时候推算不对。例如地理位置信息
当类型如果设置不对时，会导致一些功能无法正常运行，例如 Range 查询

JSON 类型	Elasticsearch 类型
字符串	1 匹配日期格式设置成 Date
	2 设置数字设置为 float 或者 long，该选项默认关闭
	3 设置为 Text, 并增加 keyword 子字段
布尔值	boolean
浮点数	float
整数	long
对象	Object
数组	由第一个非空数值的类型所决定
空值	忽略

//写入文档
PUT mapping_test/_doc/1
{
  "firstName":"Lee",
  "lastName":"Crazy",
  "loginDate":"2019-10-22T21:08:48"
}
//查看Mapping 文件
GET mapping_test/_mapping
{
  "mapping_test" : {
    "mappings" : {
      "properties" : {
        "firstName" : {
          "type" : "text",
          "fields" : {
            "keyword" : {
              "type" : "keyword",
              "ignore_above" : 256
            }
          }
        },
        "lastName" : {
          "type" : "text",
          "fields" : {
            "keyword" : {
              "type" : "keyword",
              "ignore_above" : 256
            }
          }
        },
        "loginDate" : {
          "type" : "date"
        }
      }
    }
  }
}

//dynamic mapping 推断字符的类型
PUT mapping_test/_doc/1
{
  "uid":"123",
  "isVip": false,
  "isAdmin":"true",
  "age": 18,
  "heigh" : 180
}
//返回结果
{
  "mapping_test" : {
    "mappings" : {
      "properties" : {
        "age" : {
          "type" : "long"
        },
        "heigh" : {
          "type" : "long"
        },
        "isAdmin" : {
          "type" : "text",
          "fields" : {
            "keyword" : {
              "type" : "keyword",
              "ignore_above" : 256
            }
          }
        },
        "isVip" : {
          "type" : "boolean"
        },
        "uid" : {
          "type" : "text",
          "fields" : {
            "keyword" : {
              "type" : "keyword",
              "ignore_above" : 256
            }
          }
        }
      }
    }
  }
}

显示 Mapping 设置与常见参数

控制当前字段是否被索引

index - 控制当前字段是否被索引。默认为 true。如果设置成 false，该字段不可被搜索。

PUT users
{
  "mappings" : {
    "properties" : {
      "firstName" : {
        "type" : "text"
      },
      "lastName" : {
        "type" : "text"
      },
      "mobile" : {
        "type" : "text",
        "index": false
      }
    }
  }
}

null_value

需要对 NULL 值实现搜索
只有 Keyword 类型支持设定 Null_Value

DELETE users
PUT users
{
  "mappings" : {
    "properties" : {
      "firstName" : {
        "type" : "text"
      },
      "lastName" : {
        "type" : "text"
      },
      "mobile" : {
        "type" : "keyword", //这个如果是text 无法设置为空
        "null_value": "NULL"
      }
    }
  }
}
PUT users/_doc/2
{
"firstName":"Li",
"lastName": "Sunke",
"mobile": null
}
GET users/_search?q=mobile:NULL
//搜索结果
"_source" : {
        "firstName" : "Li",
        "lastName" : "Sunke",
        "mobile" : null
      }

copy_to

_all 在 7 中已经被 copy_to 所替代
满足一些特定的搜索需求
copy_to 将字段的数值拷贝到目标字段，实现类似 _all 的作用
copy_to 的目标字段不出现在_source 中

DELETE users
PUT users
{
"mappings": {
  "properties": {
    "firstName":{
      "type": "text",
      "copy_to": "fullName"
    },
    "lastName":{
      "type": "text",
      "copy_to": "fullName"
    }
  }
}
}
PUT users/_doc/1
{
  "firstName":"Li",
  "lastName": "Sunke"
}
//没有新建字段
GET users/_doc/1
{
  "_index" : "users",
  "_type" : "_doc",
  "_id" : "1",
  "_version" : 1,
  "_seq_no" : 0,
  "_primary_term" : 1,
  "found" : true,
  "_source" : {
    "firstName" : "Li",
    "lastName" : "Sunke"
  }
}
GET users/_search?q=fullName:(Li sunke)

专题目录

ElasticStack-安装篇
 ElasticStack-elasticsearch篇
 ElasticStack-logstash篇
 elasticSearch-mapping相关
 elasticSearch-分词器介绍
 elasticSearch-分词器实践笔记
 elasticSearch-同义词分词器自定义实践
 docker-elk集群实践
 filebeat与logstash实践
 filebeat之pipeline实践
 Elasticsearch 7.x 白金级破解实践
 elk的告警调研与实践

elasticSearch-分词器介绍

发表于 2021-06-08 更新于 2024-11-02 阅读次数：

Anaiysis 与 Analyzer

Analysis - 文本分析是把全文本转换成一系列的单词（term /token）的过程，也叫分词
Analysis 是通过 Analyzer 来实现的
- 可使用 Elasticesearch 内置的分析器或者按需求定制化分析器
除了在数据写入时转换词条，匹配 Query 语句时候也需要用相同的分析器会查询语句进行分析

Analyzer 的组成

分词器是专门处理分词的组件，Analyzer 由三部分组成
- Character Filters （针对原始文本处理，例如去除 html）
- Tokenizer（按照规则切分为单词）
- Token Filter （将切分的单词进行加工，小写，删除 stopwords，增加同义语）

V22weN

Elasticsearch 的内置分词器

Standard Analyzer - 默认分词器，按词切分，小写处理
Simple Analyzer - 按照非字母切分（符号被过滤），小写处理
Stop Analyzer - 小写处理，停用词过滤（the ，a，is）
Whitespace Analyzer - 按照空格切分，不转小写
Keyword Analyzer - 不分词，直接将输入当做输出
Patter Analyzer - 正则表达式，默认 \W+
Language - 提供了 30 多种常见语言的分词器
Customer Analyzer 自定义分词器

Standard Analyzer

默认的分词器
按词切分

小写处理

#standard
GET _analyze
{
"analyzer": "standard",
"text": "2 running Quick brown-foxes leap over lazy dogs in the summer evening."
}

Simple Analyzer

按照非字母切分，非字母的都被去除

小写处理

#simple 去除非字母的 ：2 -  xi
GET _analyze
{
"analyzer": "simple",
"text": "2 running Quick brown-foxes leap over lazy dogs in the summer evening."
}

Whitespace Analyzer

空格切分

#stop
GET _analyze
{
"analyzer": "whitespace",
"text": "2 running Quick brown-foxes leap over lazy dogs in the summer evening."
}

Stop Analyzer

相比 Simple Analyzer
多了 stop filter
- 后把 the ，a, is,in 等修饰性词语去除

GET _analyze
{
"analyzer": "stop",
"text": "2 running Quick brown-foxes leap over lazy dogs in the summer evening."
}

OZVFq4

Keyword Analyzer

不分词，直接将输入当作一个 term 输出

#keyword
GET _analyze
{
"analyzer": "keyword",
"text": "2 running Quick brown-foxes leap over lazy dogs in the summer evening."
}

Pattern Analyzer

通过正则表达进行分词

默认是 \W+，非字符的符号进行分隔

GET _analyze
{
"analyzer": "pattern",
"text": "2 running Quick brown-foxes leap over lazy dogs in the summer evening."
}

Language Analyzer

各国语言分词

#english
GET _analyze
{
"analyzer": "english",
"text": "2 running Quick brown-foxes leap over lazy dogs in the summer evening."
}

使用 _analyzer Api

直接指定 Analyzer 进行测试

GET _analyze
{
  "analyzer": "standard",
  "text" : "Mastering Elasticsearch , elasticsearch in Action"
}
//返回结果
{
  "tokens" : [
    {
      "token" : "mastering",
      "start_offset" : 0,
      "end_offset" : 9,
      "type" : "<ALPHANUM>",
      "position" : 0
    },
    {
      "token" : "elasticsearch",
      "start_offset" : 10,
      "end_offset" : 23,
      "type" : "<ALPHANUM>",
      "position" : 1
    },
    {
      "token" : "elasticsearch",
      "start_offset" : 26,
      "end_offset" : 39,
      "type" : "<ALPHANUM>",
      "position" : 2
    },
    {
      "token" : "in",
      "start_offset" : 40,
      "end_offset" : 42,
      "type" : "<ALPHANUM>",
      "position" : 3
    },
    {
      "token" : "action",
      "start_offset" : 43,
      "end_offset" : 49,
      "type" : "<ALPHANUM>",
      "position" : 4
    }
  ]
}

指定索引的字段进行测试

POST books/_analyze
{
"field": "title",
"text": "Mastering Elasticesearch"
}

自定义分词进行测试

POST /_analyze
{
  "tokenizer": "standard", 
  "filter": ["lowercase"],
  "text": "Mastering Elasticesearch"
}
//结果返回
{
  "tokens" : [
    {
      "token" : "mastering",
      "start_offset" : 0,
      "end_offset" : 9,
      "type" : "<ALPHANUM>",
      "position" : 0
    },
    {
      "token" : "elasticesearch",
      "start_offset" : 10,
      "end_offset" : 24,
      "type" : "<ALPHANUM>",
      "position" : 1
    }
  ]
}

专题目录

elasticSearch-同义词分词器自定义实践

发表于 2021-06-04 更新于 2024-11-02 阅读次数：

概述

在英语中，一个单词常常是另一个单词的“变种”，如：happy=>happiness，这里的变形就是处理单复数，happy叫做happiness的词干（stem）。而adult=>man,woman，是处理同义词。

或者再如下面，需要达到搜索都能搜索出来，达到一定精确度。

裙子,裙
西红柿,番茄
china,中国,中华人民共和国
男生,男士,man
女生,女士,women

于是就有了需要自定义分词器解决同义词的场景。

实践自定义分词器

自定义分词器其实也就是组合

Character Filter
Tokenizer
Token Filter
这三个的过程。默认分词器仅仅是把这3个默认组合了。

Character Filters

在 Tokenizer 之前对文本进行处理，例如增加删除及替换字符。可以配置多个 Character Filters。会影响 Tokenizer 的 position 和 offset 信息
一些自带的 Character Filters
- HTML strip - 去除 html 标签
- Mapping - 字符串替换
- Pattern replace - 正则匹配替换

Demo char_filter

html_strip 去除html标签

POST _analyze
{
"tokenizer":"keyword",
"char_filter":["html_strip"],
"text": "<b>hello world</b>"
}
//结果
{
"tokens" : [
  {
    "token" : "hello world",
    "start_offset" : 3,
    "end_offset" : 18,
    "type" : "word",
    "position" : 0
  }
]
}

mapping 字符串替换

POST _analyze
{
"tokenizer": "standard",
"char_filter": [
    {
      "type" : "mapping",
      "mappings" : [ "- => _"]
    }
  ],
"text": "123-456, I-test! test-990 650-555-1234"
}
//返回
{
"tokens" : [
  {
    "token" : "123_456",
    "start_offset" : 0,
    "end_offset" : 7,
    "type" : "<NUM>",
    "position" : 0
  },
  {
    "token" : "I_test",
    "start_offset" : 9,
    "end_offset" : 15,
    "type" : "<ALPHANUM>",
    "position" : 1
  },
  {
    "token" : "test_990",
    "start_offset" : 17,
    "end_offset" : 25,
    "type" : "<ALPHANUM>",
    "position" : 2
  },
  {
    "token" : "650_555_1234",
    "start_offset" : 26,
    "end_offset" : 38,
    "type" : "<NUM>",
    "position" : 3
  }
]
}

pattern_replace 正则表达式

GET _analyze
{
"tokenizer": "standard",
"char_filter": [
    {
      "type" : "pattern_replace",
      "pattern" : "http://(.*)",
      "replacement" : "$1"
    }
  ],
  "text" : "http://www.elastic.co"
}
//返回
{
"tokens" : [
  {
    "token" : "www.elastic.co",
    "start_offset" : 0,
    "end_offset" : 21,
    "type" : "<ALPHANUM>",
    "position" : 0
  }
]
}

Tokenizer

将原始的文本按照一定的规则，切分为词（term or token）
Elasticsearch 内置的 Tokenizers
- whitespace | standard | uax_url_email | pattern | keyword | path hierarchy

Demo tokenizer

path_hierarchy 通过路劲切分

POST _analyze
{
"tokenizer":"path_hierarchy",
"text":"/user/ymruan/a"
}
{
"tokens" : [
  {
    "token" : "/user",
    "start_offset" : 0,
    "end_offset" : 5,
    "type" : "word",
    "position" : 0
  },
  {
    "token" : "/user/ymruan",
    "start_offset" : 0,
    "end_offset" : 12,
    "type" : "word",
    "position" : 0
  },
  {
    "token" : "/user/ymruan/a",
    "start_offset" : 0,
    "end_offset" : 14,
    "type" : "word",
    "position" : 0
  }
]
}

Token tokenizer Filter

将 Tokenizer 输出的单词，进行增加、修改、删除
自带的 Token Filters
- Lowercase |stop| synonym（添加近义词）

Demo filter

whitespace 空格

GET _analyze
{
"tokenizer": "whitespace", 
"filter": ["stop","snowball"], //on the a
"text": ["The gilrs in China are playing this game!"]
}
{
"tokens" : [
  {
    "token" : "The", //大写的The 不做过滤
    "start_offset" : 0,
    "end_offset" : 3,
    "type" : "word",
    "position" : 0
  },
  {
    "token" : "gilr",
    "start_offset" : 4,
    "end_offset" : 9,
    "type" : "word",
    "position" : 1
  },
  {
    "token" : "China",
    "start_offset" : 13,
    "end_offset" : 18,
    "type" : "word",
    "position" : 3
  },
  {
    "token" : "play",
    "start_offset" : 23,
    "end_offset" : 30,
    "type" : "word",
    "position" : 5
  },
  {
    "token" : "game!",
    "start_offset" : 36,
    "end_offset" : 41,
    "type" : "word",
    "position" : 7
  }
]
}

自定义 analyzer

官网自定义分词器的标准格式

自定义分析器标准格式是：
PUT /my_index
{
  "settings": {
      "analysis": {
          "char_filter": { ... custom character filters ... },//字符过滤器
          "tokenizer": { ... custom tokenizers ... },//分词器
          "filter": { ... custom token filters ... }, //词单元过滤器
          "analyzer": { ... custom analyzers ... }
      }
  }
}

自定义分词器

#定义自己的分词器
PUT my_index
{
"settings": {
  "analysis": {
    "analyzer": {
      "my_custom_analyzer":{
        "type":"custom",
        "char_filter":[
          "emoticons"
        ],
        "tokenizer":"punctuation",
        "filter":[
          "lowercase",
          "english_stop"
        ]
      }
    },
    "tokenizer": {
      "punctuation":{
        "type":"pattern",
        "pattern": "[ .,!?]"
      }
    },
    "char_filter": {
      "emoticons":{
        "type":"mapping",
        "mappings" : [ 
          ":) => happy",
          ":( => sad"
        ]
      }
    },
    "filter": {
      "english_stop":{
        "type":"stop",
        "stopwords":"_english_"
      }
    }
  }
}
}

自定义同义词需求解决

synonym_graph
我们定义一个my_synonym_filter的filter进行处理同义词,
同时自定义自己的分词器my_custom_analyzer,并指定字段title使用my_custom_analyzer分词器

PUT my_index
{
  "settings": {
    "analysis": {
      "analyzer": {
        "my_custom_analyzer": {
          "type": "custom",
          "tokenizer": "standard",
          "filter": [
            "lowercase",
            "my_synonym_filter"
          ]
        }
      },
      "filter": {
        "my_synonym_filter": {
          "type": "synonym",
          "synonyms": [
            "british,english",
            "queen,monarch"
          ]
        }
      }
    }
  },
  "mappings": {
    "properties": {
      "title": {
        "type": "text",
        "analyzer": "my_custom_analyzer",
        "search_analyzer": "my_custom_analyzer"
      },
      "author": {
        "type": "keyword"
      }
    }
  }
}

测试一下分词器的效果

POST my_index/_analyze
{
  "field": "title",
  "text": "Elizabeth is the English queen",
  "analyzer": "my_custom_analyzer"
}

我们会发现Elizabeth is the English queen，包含了english，而我们设置了british,english,为同义词，所以分词器就包含了british,english

{
  "tokens" : [
    {
      "token" : "elizabeth",
      "start_offset" : 0,
      "end_offset" : 9,
      "type" : "<ALPHANUM>",
      "position" : 0
    },
    {
      "token" : "is",
      "start_offset" : 10,
      "end_offset" : 12,
      "type" : "<ALPHANUM>",
      "position" : 1
    },
    {
      "token" : "the",
      "start_offset" : 13,
      "end_offset" : 16,
      "type" : "<ALPHANUM>",
      "position" : 2
    },
    {
      "token" : "english",
      "start_offset" : 17,
      "end_offset" : 24,
      "type" : "<ALPHANUM>",
      "position" : 3
    },
    {
      "token" : "british",
      "start_offset" : 17,
      "end_offset" : 24,
      "type" : "SYNONYM",
      "position" : 3
    },
    {
      "token" : "queen",
      "start_offset" : 25,
      "end_offset" : 30,
      "type" : "<ALPHANUM>",
      "position" : 4
    },
    {
      "token" : "monarch",
      "start_offset" : 25,
      "end_offset" : 30,
      "type" : "SYNONYM",
      "position" : 4
    }
  ]
}

我们再通过插入数据来测试一下 queen,monarch 同义词


PUT my_index/_doc/1
{
  "title": "Elizabeth is the English queen"
}


GET my_index/_search
{
  "query": {
    "match": {
      "title": "monarch"
    }
  }
}

我们发现我们能通过monarch 查询出来含queen的数据

{
  "took" : 3,
  "timed_out" : false,
  "_shards" : {
    "total" : 1,
    "successful" : 1,
    "skipped" : 0,
    "failed" : 0
  },
  "hits" : {
    "total" : {
      "value" : 1,
      "relation" : "eq"
    },
    "max_score" : 0.2876821,
    "hits" : [
      {
        "_index" : "my_index2",
        "_type" : "_doc",
        "_id" : "1",
        "_score" : 0.2876821,
        "_source" : {
          "title" : "Elizabeth is the English queen"
        }
      }
    ]
  }
}

自此解决同义词的需求

动态设置同义词的方案

开启热更新，仅仅适用search_analyzer
https://www.elastic.co/guide/en/elasticsearch/reference/7.x/indices-reload-analyzers.html

PUT my_index
{
  "settings": {
    "analysis": {
      "analyzer": {
        "my_custom_analyzer": {
          "type": "custom",
          "tokenizer": "standard",
          "filter": [
            "lowercase",
            "word_syn"
          ]
        }
      },
      "filter": {
        "word_syn": {
          "type": "synonym_graph",
          "synonyms_path": "analysis/synonym.txt",
          ## 开启热更新
          "updateable": true  
        },
      }
    }
  },
  "mappings": {
    "properties": {
      "title": {
        "type": "text",
        "analyzer": "english",
        "search_analyzer": "my_custom_analyzer"
      },
      "author": {
        "type": "keyword"
      }
    }
  }
}

热更新重载分词器

1	POST my_index/_reload_search_analyzers

专题目录

elasticSearch-分词器实践笔记

发表于 2021-06-04 更新于 2024-11-02 阅读次数：

概述

1、为什么命名有包含搜索关键词的文档，但结果里面就没有相关文档呢？
2、我存进去的文档到底被分成哪些词(term)了？

带着这些问题，我们来实践一下

实践

测试一下

让我们从一个实例出发，如下创建一个文档：

PUT test/_doc/1
{
  "msg":"Eating an apple a day keeps doctor away"
}

然后我们做一个查询，我们试图通过搜索 eat这个关键词来搜索这个文档

POST test/_search
{
  "query":{
    "match":{
      "msg":"eat"
    }
  }
}

ES的返回结果为0。这不太对啊，我们用最基本的字符串查找也应该能匹配到上面新建的文档才对啊！

分词原理

搜索引擎的核心是倒排索引，而倒排索引的基础就是分词。所谓分词可以简单理解为将一个完整的句子切割为一个个单词的过程。在 es 中单词对应英文为 term。我们简单看个例子：
KeMwjZ

ES 的倒排索引即是根据分词后的单词创建，即 我、爱、北京、天安门这4个单词。这也意味着你在搜索的时候也只能搜索这4个单词才能命中该文档。

实际上 ES 的分词不仅仅发生在文档创建的时候，也发生在搜索的时候，如下图所示：

FKOsVy

读时分词发生在用户查询时，ES 会即时地对用户输入的关键词进行分词，分词结果只存在内存中，当查询结束时，分词结果也会随即消失。而写时分词发生在文档写入时，ES 会对文档进行分词后，将结果存入倒排索引，该部分最终会以文件的形式存储于磁盘上，不会因查询结束或者 ES 重启而丢失。

ES 中处理分词的部分被称作分词器，英文是Analyzer，它决定了分词的规则。ES 自带了很多默认的分词器，比如Standard、 Keyword、Whitespace等等，默认是 Standard。当我们在读时或者写时分词时可以指定要使用的分词器。

写时分词

回到上手阶段，我们来看下写入的文档最终分词结果是什么。通过如下 api 可以查看：

POST test/_analyze
{
  "field": "msg",
  "text": "Eating an apple a day keeps doctor away"
}

其中 test为索引名，_analyze 为查看分词结果的 endpoint，请求体中 field 为要查看的字段名，text为具体值。该 api 的作用就是请告诉我在 test 索引使用 msg 字段存储一段文本时，es 会如何分词。

返回结果如下:

{
  "tokens": [
    {
      "token": "eating",
      "start_offset": 0,
      "end_offset": 6,
      "type": "<ALPHANUM>",
      "position": 0
    },
    {
      "token": "an",
      "start_offset": 7,
      "end_offset": 9,
      "type": "<ALPHANUM>",
      "position": 1
    },
    {
      "token": "apple",
      "start_offset": 10,
      "end_offset": 15,
      "type": "<ALPHANUM>",
      "position": 2
    },
    {
      "token": "a",
      "start_offset": 16,
      "end_offset": 17,
      "type": "<ALPHANUM>",
      "position": 3
    },
    {
      "token": "day",
      "start_offset": 18,
      "end_offset": 21,
      "type": "<ALPHANUM>",
      "position": 4
    },
    {
      "token": "keeps",
      "start_offset": 22,
      "end_offset": 27,
      "type": "<ALPHANUM>",
      "position": 5
    },
    {
      "token": "doctor",
      "start_offset": 28,
      "end_offset": 34,
      "type": "<ALPHANUM>",
      "position": 6
    },
    {
      "token": "away",
      "start_offset": 35,
      "end_offset": 39,
      "type": "<ALPHANUM>",
      "position": 7
    }
  ]
}

返回结果中的每一个 token即为分词后的每一个单词，我们可以看到这里是没有 eat 这个单词的，这也解释了在上手中我们搜索 eat 没有结果的情况。如果你去搜索 eating ，会有结果返回。

写时分词器需要在 mapping 中指定，而且一经指定就不能再修改，若要修改必须新建索引。如下所示我们新建一个名为ms_english 的字段，指定其分词器为 english：

PUT test/_doc/_mapping?include_type_name=true
{
  "properties": {
    "msg_english":{
      "type":"text",
      "analyzer": "english"
    }
  }
}

读时分词

由于读时分词器默认与写时分词器默认保持一致，拿上手中的例子，你搜索 msg 字段，那么读时分词器为 Standard ，搜索 msg_english 时分词器则为 english。这种默认设定也是非常容易理解的，读写采用一致的分词器，才能尽最大可能保证分词的结果是可以匹配的。
然后 ES 允许读时分词器单独设置，如下所示：

POST test/_search
  {
    "query":{
      "match":{
        "msg":{
          "query": "eating",
          "analyzer": "english"
        }
      }
    }
  }

如上 analyzer 字段即可以自定义读时分词器，一般来讲不需要特别指定读时分词器。

如果不单独设置分词器，那么读时分词器的验证方法与写时一致；如果是自定义分词器，那么可以使用如下的 api 来自行验证结果。

POST _analyze
{
  "text":"eating",
  "analyzer":"english"
}

{
  "tokens": [
    {
      "token": "eat",
      "start_offset": 0,
      "end_offset": 6,
      "type": "<ALPHANUM>",
      "position": 0
    }
  ]
}

由上可知 english分词器会将 eating处理为 eat，而standard则不作处理，即eating处理为eating。

通过下图的分析，我们就知道为什么我们查不出数据了。
UxvCpN

解决问题

由于 eating只是 eat的一个变形，我们依然希望输入 eat时可以匹配包含 eating的文档，那么该如何解决呢？
答案很简单，既然原因是在分词结果不匹配，那么我们就换一个分词器呗~ 我们可以先试下 ES 自带的 english分词器，如下：

# 增加字段 msg_english，与 msg 做对比
PUT test/_doc/_mapping?include_type_name=true
{
  "properties": {
    "msg_english":{
      "type":"text",
      "analyzer": "english"
    }
  }
}

# 写入相同文档
PUT test/doc/1
{
  "msg":"Eating an apple a day keeps doctor away",
  "msg_english":"Eating an apple a day keeps doctor away"
}

# 搜索 msg_english 字段
POST test/_search
{
  "query": {
    "match": {
      "msg_english": "eat"
    }
  }
}

执行上面的内容，我们会发现结果有内容了，原因也很简单，如下图所示：
Ctf9xM
由上图可见 english分词器会将 eating分词为 eat，此时我们搜索 eat或者 eating肯定都可以匹配对应的文档了。至此，需求解决。

自定义分词器

假设有一天产品提了一个需求：西红柿,番茄的同义词都要能搜索出来,这个时候我们就可以自定义分词器。

elasticSearch-同义词实践

专题目录

Flink CDC同步mysql和pg关联数据到es实践

发表于 2021-05-28 更新于 2024-11-02 阅读次数：

xyjcvD

概述

最近有需求要将商品同步到es，并做pv、uv的宽表处理，评估一下Flink CDC的能力，实践一下Flink CDC的同步功能。
本次目标是将mysql和pg的表数据，在es进行宽表处理。

本次模拟场景产品表products、订单表orders在mysql数据库，物流表shipments在postgres数据库，最终的宽表enriched_orders在elasticsearch.

模拟电商公司的订单表和物流表，需要对订单数据进行统计分析，对于不同的信息需要进行关联后续形成订单的大宽表后，交给下游的业务方使用 ES 做数据分析，这个案例演示了如何只依赖 Flink 不依赖其他组件，借助 Flink 强大的计算能力实时把 Binlog 的数据流关联一次并同步至 ES 。

版本

Apache Flink 1.13.1
flink-sql-connector-elasticsearch7_2.11-1.13.0.jar
flink-sql-connector-mysql-cdc-1.4.0.jar
flink-sql-connector-postgres-cdc-1.4.0.jar
java8

安装

docker方式安装

1 2	FLINK_PROPERTIES="jobmanager.rpc.address: jobmanager" docker network create flink-network

JobManager

docker run \
--rm \
--name=jobmanager \
--network flink-network \
--publish 8081:8081 \
--env FLINK_PROPERTIES="${FLINK_PROPERTIES}" \
flink:1.13.1-scala_2.11 jobmanager

TaskManager

docker run \
--rm \
--name=taskmanager \
--network flink-network \
--env FLINK_PROPERTIES="${FLINK_PROPERTIES}" \
flink:1.13.1-scala_2.11 taskmanager

下载方式安装

https://flink.apache.org/zh/downloads.html

下载最新版本即可。

Flink sql常见命令

// 停止集群
bin/stop-cluster.sh
// 启动集群
bin/start-cluster.sh
// 进入flink sql 客户端命令行界面
bin/sql-client.sh embedded
// 查看当前运行的jobs
bin/flink list
// 查看所有的任务，包括失败、成功、取消的
bin/flink list -a
// 取消命令
bin/flink cancel jobID

实践

1、下载 docker-compose.yml

先使用docker创造一些mysql 和 pg的数据

version: '2.1'
services:
  postgres:
    image: debezium/example-postgres:1.1
    ports:
      - "5432:5432"
    environment:
      - POSTGRES_PASSWORD=1234
      - POSTGRES_DB=postgres
      - POSTGRES_USER=postgres
      - POSTGRES_PASSWORD=postgres
  mysql:
    image: debezium/example-mysql:1.1
    ports:
      - "3306:3306"
    environment:
      - MYSQL_ROOT_PASSWORD=123456
      - MYSQL_USER=mysqluser
      - MYSQL_PASSWORD=mysqlpw
  elasticsearch:
    image: elastic/elasticsearch:7.6.0
    environment:
      - cluster.name=docker-cluster
      - bootstrap.memory_lock=true
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
      - discovery.type=single-node
    ports:
      - "9200:9200"
      - "9300:9300"
    ulimits:
      memlock:
        soft: -1
        hard: -1
      nofile:
        soft: 65536
        hard: 65536
  kibana:
    image: elastic/kibana:7.6.0
    ports:
      - "5601:5601"
  zookeeper:
    image: wurstmeister/zookeeper:3.4.6
    ports:
      - "2181:2181"
  kafka:
    image: wurstmeister/kafka:2.12-2.2.1
    ports:
      - "9092:9092"
      - "9094:9094"
    depends_on:
      - zookeeper
    environment:
      - KAFKA_ADVERTISED_LISTENERS=INSIDE://:9094,OUTSIDE://localhost:9092
      - KAFKA_LISTENERS=INSIDE://:9094,OUTSIDE://:9092
      - KAFKA_LISTENER_SECURITY_PROTOCOL_MAP=INSIDE:PLAINTEXT,OUTSIDE:PLAINTEXT
      - KAFKA_INTER_BROKER_LISTENER_NAME=INSIDE
      - KAFKA_ZOOKEEPER_CONNECT=zookeeper:2181
      - KAFKA_CREATE_TOPICS="user_behavior:1:1"
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock

2、进入 mysql 容器，初始化数据：

1	docker-compose exec mysql mysql -uroot -p123456

-- MySQL
CREATE DATABASE mydb;
USE mydb;
CREATE TABLE products (
  id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,
  name VARCHAR(255) NOT NULL,
  description VARCHAR(512)
);
ALTER TABLE products AUTO_INCREMENT = 101;

INSERT INTO products
VALUES (default,"scooter","Small 2-wheel scooter"),
       (default,"car battery","12V car battery"),
       (default,"12-pack drill bits","12-pack of drill bits with sizes ranging from #40 to #3"),
       (default,"hammer","12oz carpenter's hammer"),
       (default,"hammer","14oz carpenter's hammer"),
       (default,"hammer","16oz carpenter's hammer"),
       (default,"rocks","box of assorted rocks"),
       (default,"jacket","water resistent black wind breaker"),
       (default,"spare tire","24 inch spare tire");

CREATE TABLE orders (
  order_id INTEGER NOT NULL AUTO_INCREMENT PRIMARY KEY,
  order_date DATETIME NOT NULL,
  customer_name VARCHAR(255) NOT NULL,
  price DECIMAL(10, 5) NOT NULL,
  product_id INTEGER NOT NULL,
  order_status BOOLEAN NOT NULL -- 是否下单
) AUTO_INCREMENT = 10001;

INSERT INTO orders
VALUES (default, '2020-07-30 10:08:22', 'Jark', 50.50, 102, false),
       (default, '2020-07-30 10:11:09', 'Sally', 15.00, 105, false),
       (default, '2020-07-30 12:00:30', 'Edward', 25.25, 106, false);

3、进入postgres 容器，初始化数据：

1	docker-compose exec postgres psql -h localhost -U postgres

-- PG
CREATE TABLE shipments (
  shipment_id SERIAL NOT NULL PRIMARY KEY,
  order_id SERIAL NOT NULL,
  origin VARCHAR(255) NOT NULL,
  destination VARCHAR(255) NOT NULL,
  is_arrived BOOLEAN NOT NULL
);
ALTER SEQUENCE public.shipments_shipment_id_seq RESTART WITH 1001;
ALTER TABLE public.shipments REPLICA IDENTITY FULL;

INSERT INTO shipments
VALUES (default,10001,'Beijing','Shanghai',false),
       (default,10002,'Hangzhou','Shanghai',false),
       (default,10003,'Shanghai','Hangzhou',false);

4、下载以下 jar 包到 <FLINK_HOME>/lib/:
查看上方相关链接下载

--FlinkSQL
CREATE TABLE products (
  id INT,
  name STRING,
  description STRING
) WITH (
  'connector' = 'mysql-cdc',
  'hostname' = 'localhost',
  'port' = '3306',
  'username' = 'root',
  'password' = '123456',
  'database-name' = 'mydb',
  'table-name' = 'products'
);

CREATE TABLE orders (
  order_id INT,
  order_date TIMESTAMP(0),
  customer_name STRING,
  price DECIMAL(10, 5),
  product_id INT,
  order_status BOOLEAN
) WITH (
  'connector' = 'mysql-cdc',
  'hostname' = 'localhost',
  'port' = '3306',
  'username' = 'root',
  'password' = '123456',
  'database-name' = 'mydb',
  'table-name' = 'orders'
);

CREATE TABLE shipments (
  shipment_id INT,
  order_id INT,
  origin STRING,
  destination STRING,
  is_arrived BOOLEAN
) WITH (
  'connector' = 'postgres-cdc',
  'hostname' = 'localhost',
  'port' = '5432',
  'username' = 'postgres',
  'password' = 'postgres',
  'database-name' = 'postgres',
  'schema-name' = 'public',
  'table-name' = 'shipments'
);

CREATE TABLE enriched_orders (
  order_id INT,
  order_date TIMESTAMP(0),
  customer_name STRING,
  price DECIMAL(10, 5),
  product_id INT,
  order_status BOOLEAN,
  product_name STRING,
  product_description STRING,
  shipment_id INT,
  origin STRING,
  destination STRING,
  is_arrived BOOLEAN,
  PRIMARY KEY (order_id) NOT ENFORCED
) WITH (
    'connector' = 'elasticsearch-7',
    'hosts' = 'http://localhost:9200',
    'index' = 'enriched_orders'
);

INSERT INTO enriched_orders
SELECT o.*, p.name, p.description, s.shipment_id, s.origin, s.destination, s.is_arrived
FROM orders AS o
LEFT JOIN products AS p ON o.product_id = p.id
LEFT JOIN shipments AS s ON o.order_id = s.order_id;

6、修改 mysql 和 postgres 里面的数据，观察 elasticsearch 里的结果。

--MySQL
INSERT INTO orders
VALUES (default, '2020-07-30 15:22:00', 'Jark', 29.71, 104, false);

--PG
INSERT INTO shipments
VALUES (default,10004,'Shanghai','Beijing',false);

--MySQL
UPDATE orders SET order_status = true WHERE order_id = 10004;

--PG
UPDATE shipments SET is_arrived = true WHERE shipment_id = 1004;

--MySQL
DELETE FROM orders WHERE order_id = 10004;

期间遇到的问题

https://github.com/ververica/flink-cdc-connectors/issues/197

Vue3 Composition API笔记

发表于 2021-05-24 更新于 2024-11-02 阅读次数：

概述

在使用nestjs梳理RBAC构架，顺便把前端框架升到vue3,详细看了下Vue3 Composition API,有点意思

Composition API

1.ref

ref可以代理字符串、数字、boolean等基本类型值
ref声明的值需要通过.value去改变
ref目的是为了引用原始类型值，但仍然可以引用非基本类型值例如对象

ref 本质也是reactive 可以简单地把 ref(1) 理解为这个样子 reactive({value: 1})

//用法一 代理基本类型
import { ref } from 'vue'

const refVal = ref(1)

const add = () => {
  refVal.value++ //值改变，视图更新
}

//用法二
const refObj = ref({ foo: 1 })

const add = () => {
 //需要通过.value去访问
  refObj.value.foo = 2 //值改变，触发视图更新
}

//用法三
//可以通过ref代理某个对象下面的值，复制修改响应式数据不影响原对象
const obj = { foo: 1 }

const refVal = ref(obj.foo)

const add = () => {
    refVal.value++
    console.log(refVal.value) //值改变，视图更新
    console.log(obj.foo) //值不变
}

2.reactive

reactive接受一个可代理的对象，但不能是字符串、数字、boolean等基本类型值
reactive不需要通过.value去访问属性值

import { reactive } from 'vue'

const obj = reactive({ foo: 1 })

const add = () => {
  obj.foo++ //值改变，视图更新
}

3.shallowReactive

用法同reactive，用于定义一个浅响应数据只代理第一层，当数据结构比较复杂时，每层都用proxy代理消耗性能

import { shallowReactive } from 'vue'

const obj = shallowReactive({ foo: { bar: 1 } })

const add = () => {
  obj.foo.bar = 2 //值改变，不触发视图更新
  obj.foo = { bar: 2 } //值改变，视图更新
}

4.readonly

用于定义一个只可读数据,接受一个Object对象

import { readonly } from 'vue'

const obj = readonly({ text: 'hi' })

const add = () => {
  obj.text = 'hello' //报错
}

5.shallowReadonly

import { shallowReadonly } from 'vue'

const obj = shallowReadonly({ foo: { bar: 1 } })

const add = () => {
  obj.foo = { bar: 2 } //报错
  obj.foo.bar = 2 //有效
}

6.toRef

创建一个ref类型数据, 并和以前的数据关联
相当于引用, 修改响应式数据会影响原始数据
第一个参数为 obj 对象；第二个参数为对象中的属性名

应用场景:如果想让响应式数据和原始的数据关联起来, 并且更新响应式数据之后还不想更新UI, 那么就可以使用toRef

//数据发生改变, 视图也不会自动更新

const obj = { foo: 1 }

//const obj = reactive({ foo: 1}) //reactive创建的对象会触发视图更新

const refVal = toRef(obj, 'foo') 

const add = () => {
    refVal.value++
    console.log(refVal.value) //值改变，视图不更新
    console.log(obj.foo) //值改变，视图不更新
}

7.toRefs

//用法一

<template>
  <p @click="add">{{ foo }}</p>
</template>

import { reactive, toRefs } from 'vue'

const obj = reactive({ foo: 1 })

return{
   ...toRefs(obj) //将obj里的每个属性转化为ref响应式数据
}

//用法二

//批量创建ref类型数据, 并和以前数据关联，不触发视图更新

import { reactive, toRefs } from 'vue'

const obj = { foo: 1, num: 1 }
//const obj = reactive({ foo: 1, num: 1 })  //reactive创建的对象会触发视图更新
   
const state = toRefs(obj)

const add = () => {
  state.foo.value = 2 
  state.num.value = 2
  console.log(state.foo.value) // 2 值改变，视图不更新
  console.log(obj.foo) // 2 值改变，视图不更新
}

8.shallowRef

这是一个浅层的 ref ,只代理.value 可用于优化性能


import { shallowRef, triggerRef } from 'vue'

const obj = shallowRef({ foo: 1 })

//shallowRef只代理 ref 对象本身，也就是说只有 .value 是被代理的，而 .value 所引用的对象并没有被代理

const add = () => {
  obj.value.foo = 2 //值改变，视图不更新
  triggerRef(obj) // 可通过修改值后立即驱动视图更新
   
  obj.value = { foo: 2 } //值改变，视图更新
 
}

9.unref

unref接收一个值，如果这个值是 ref 就返回 .value，否则原样返回

10.markRaw

markRaw 方法可以将原始数据标记为非响应式的，即使用 ref 或 reactive 将其包装，仍无法实现数据响应式，其接收一个参数，即原始数据，并返回被标记后的数据

markRaw 函数所做的事情，就是在数据对象上定义 __v_skip 属性，从而跳过代理

import { markRaw, reactive } from 'vue'

//通过markRow代理过的对象，不会触发视图更新
//markRow可用来数据改变但不需要视图改变的情况，用于提升性能。
const obj = { foo: 1 }
const obj2 = markRaw(obj)
const state = reactive(obj2)

const add = () => {
  state.foo = 2 //值改变，视图不更新
  console.log(state) //2
  console.log(obj) //2 
}

11.toRaw

toRaw方法用于拿到原始数据，对原始数据进行修改,不会更新UI界面，
与markRow()方法类似可用于提升性能，不同的是 markRow接收的不是被代理过的响应式数据
toRaw 方法是用于获取 ref 或 reactive 对象的原始数据的

import { toRaw, reactive, ref } from 'vue'

//代理reactive对象
const obj = reactive({ foo: 1 })
const obj2 = toRaw(obj)

const add = () => {
  obj2.foo = 2 //值改变，视图不会更新
}

//代理ref创建的对象

const obj = ref({ foo: 1 })
const obj2 = toRaw(obj.value) //与reactive不同的是，需要用.value去获取原始数据，因为经过Vue处理之后,.value中保存的才是当初创建时传入的那个原始数据

const add = () => {
   obj2.foo = 2 //值改变，视图不会更新
   console.log(obj.value) //输出 { foo: 2 }
}

12.isRef

用于判断数据是否是ref创建的，Vue3创建ref的时候会增加__v_isRef: true属性来标识ref数据

import { ref, isRef } from 'vue'

const val = ref(1)

console.log(isRef(val)) //true

13.isReactive

判断数据对象是否是 reactive

import { ref, isReactive } from 'vue'

const obj = reactive({ foo: 1 })

console.log(isReactive(obj)) //true

14.isReadonly

判断数据对象是否是readonly只可读

import { readonly, isReadonly } from 'vue'

const val = readonly({ foo: 1 })

console.log(isReadonly(val)) //true

15.isProxy

用于判断对象是否是reactive 或 readonly 创建的代理对象

import { readonly, reactive, isProxy } from 'vue'

const obj = reactive({ foo: 1 })
const val = readonly({ foo: 1 })

console.log(isProxy(obj)) //true
console.log(isProxy(val)) //true

16.computed

用法与Vue2中的computed一样

import { ref, computed } from 'vue'

//写法-
const val = ref(1)

//vue3中计算属性的函数中如果只传入一个回调函数，表示的是get
const doubule = computed(() => val.value * 2)

const add = () => {
  val.value = 3
  console.log(doubule.value) // 6 需要通过.value去访问
}

//写法2
<template>
  <input type="text" v-model="doubule">
</template>

const val = ref(1)

const doubule = computed(() => val.value * 2)

const doubule = computed({
  get() {
  	 //dobule的返回值
     return obj.foo * 2
  },
  set(value) {
  	 //写你的逻辑代码
     val.value++
     obj.foo = val.value
  }
})

17.watch

watch监听数据变化，需手动传入监听的数据，返回新值和旧值
与vue2不同的是 vue2需要通过computed计算才会返回新值和旧值，否则返回的都是新值。

import { ref, reactive, watch } from 'vue'
//监听ref创建的数据

const val = ref(0)

watch(val,(newVal, oldVal) => {
     console.log(newVal) //1 输出新值
     console.log(oldVal) //0 输出旧值
  },
  {
    immediate: false, //是否在初始化监听
    deep: false //是否开启深度监听
  }
)
const add = () => {
   val.value = 1 //值改变
}

//监听reactive创建的数据,与ref不同的是需要用箭头函数指向要监听的数据
const obj = reactive({ foo: 0 })

watch(() => obj.foo,(newVal, oldVal) => {
     console.log(newVal) //1 输出新值
     console.log(oldVal) //0 输出旧值
  },
  {
    immediate: false, //是否在初始化监听
    deep: false //是否开启深度监听
  }
)

const add = () => {
   obj.foo = 1 //值改变
}

//监听多个数据源
const val = ref(1)

const obj = reactive({ foo: 1 })

watch([() => obj.foo, val], ([newFoo, newVal], [oldFoo, oldVal]) => {
      console.log(newFoo, oldFoo)
      console.log(newVal, oldVal)
})

 const add = () => {
     val.value += 2
     obj.foo++
}

//watch 接受一个stop

18.watchEffect

watchEffect也是监听数据变化。
与watch不同的是:
1.不需要手动传入依赖
2.每次初始化都会执行
3.无法获取到原值，只能得到变化后的值

import { ref, reactive, watchEffect } from 'vue'

const obj = reactive({ foo: 1 })

const val = ref(0)

watchEffect(() => {
      console.log(obj.foo)
      console.log(val.value)
})

const add = () => {
  val.value++
  obj.foo++ 
}

//watchEffect还接受一个函数作为参数 ，可用于清除副作用
watchEffect(async () => {
    const data = await fetch(obj.foo)
})

//当 obj.foo 变化后，意味着将会再次发送请求，那么之前的请求怎么办呢？是否应该将之前的请求标记为 invalidate

watchEffect(async (onInvalidate) => {
    let validate = true
    onInvalidate(() => {
        validate = false
    })
    const data = await fetch(obj.foo)
    if (validate){
        /* 正常使用 data */
    } else {
        /* 说明当前副作用已经无效了，抛弃即可 */
    }
})

19.defineComponent & PropType

两者都是为了更好的推断TS类型

import { defineComponent, PropType } from 'vue'

interface Mylist {
  name: string
  age: number
}

export default defineComponent({
  props: {
    list: Object as PropType<Mylist[]>
  },
  
  setup(){}
})

20.生命周期函数

improt{
  onBeforeMount
  onMounted
  onBeforeUpdate
  onUpdated
  onBeforeUnmount
  onUnmounted
  onActivated
  onDeactivated
  onErrorCaptured
} from 'vue'
vue3 新增的两个钩子

onRenderTracked
onRenderTriggered

export default {
  onRenderTriggered(e) {
    debugger
    // 检查哪个依赖项导致组件重新呈现
  }
}

21.customRef

自定义 ref，常用来定义需要异步获取的响应式数据

//可以用customRef实现一个搜索框防抖

<template>
  <input type="text" v-model="text">
</template>


const useDebouncedRef = (value: string, delay = 1000) => {
      let timeout: any
      /**
       * customRef回调接受两个参数
       * track用于追踪依赖
       * trigger用于触发响应
       * 回调需返回一个包含get和set方法的对象
       */
      return customRef((track, trigger) => {
        return {
          get() {
            track() //追踪该数据
            return value
          },
          set(newVal: string) {
            clearTimeout(timeout)
            timeout = setTimeout(() => {
              value = newVal
              trigger() // 数据被修改，更新ui界面
            }, delay)
         }
      }
    })
 }
const text = useDebouncedRef('')

watch(text, async (newText) => {
      if (!newText) return void 0
      console.log(newText) //停止输入1秒后输出。
})

return{ text }

22.defineProps & defineEmit

//在setup中直接用于接受props和emit，可做ts类型推导
<script setup lang="ts">
import { defineProps, defineEmit } from 'vue'

const props = defineProps<{
  foo: number,
  age?: number,
}>()

//当有多个事件且参数不同时
const emit = defineEmit<{
  (e: 'close', id: number): void
  (e: 'show', name: string, age: number): void
}>()

emit('close', 1)
emit('show', '1', 2)

//参数相同时
const emit = defineEmit<(e: 'close' | 'show', id: number) => void>()

emit('close', 1)
emit('show', 1)
</script>

23.defineAsyncComponent

//用于引入组件

<script setup lang="ts">
import { defineAsyncComponent } from 'vue'
const AsyncShow = defineAsyncComponent(
  () => import('@/components/AsyncShow.vue')
)
</script>

24.script vars

支持将组件状态驱动的 CSS 变量注入到“单个文件组件”样式中。

<template>
	//颜色改变
  <p class="text">hello</p>
</template>
<script setup lang="ts">
	const color = '#3b6af9'
</script>

<style lang="scss" scoped>
.text {
  color: v-bind(color);
}
</style>

25.provide && inject

与 Vue2中的 provide 和 inject 作用相同，只不过在Vue3中需要手动从 vue 中导入
这里简单说明一下这两个方法的作用：

provide ：向子组件以及子孙组件传递数据。接收两个参数，第一个参数是 key，即数据的名称；第二个参数为 value，即数据的值
inject ：接收父组件或祖先组件传递过来的数据。接收一个参数 key，即父组件或祖先组件传递的数据名称

// A.vue
<script>
import {provide} from 'vue'
export default {
    setup() {
        const obj= {
            name: '前端印象',
            age: 22
        }
        // 向子组件以及子孙组件传递名为info的数据
        provide('info', obj)
    }
}
</script>
// B.vue
<script>
import {inject} from 'vue'
export default {
    setup() {   
        // 接收A.vue传递过来的数据
        inject('info')  // {name: '前端印象', age: 22}
    }
}
</script>
// C.vue
<script>
import {inject} from 'vue'
export default {
    setup() {   
        // 接收A.vue传递过来的数据
        inject('info')  // {name: '前端印象', age: 22}
    }
}
</script>

26.getCurrentInstance

获取当前实例,和vue2中的this相同，用于setup函数中(不建议使用)

import { getCurrentInstance } from 'vue'
const { ctx } = getCurrentInstance()
console.log(ctx)

27.vue-router里的hooks

import { useRoute, useRouter } from 'vue-router'
const route = useRoute()
const router = useRouter()
 
console.log(route.params.id)
router.push('/xxx/xxx')

28.vuex 里的 hooks

import { useStore } from 'vue-router'

const store = useStore()

背景

问题还原

问题还原2

mongodb副本集环境搭建

bash 脚本

启动 Docker

配置副本集

相关副本集命令

事务实践

如何通过mongoose连接副本集

readPreference

mongoose事务回滚

相关链接

前言

笔记原因

others

总结

番外篇

前言

virtual使用

合并字段

virtual关联查询(子查父)

前言

populate

aggregate

对比

总结

关联技巧一：关联引用子查父

关联技巧二：关联数组转对象

字段的数据类型

Dynamic Mapping

显示 Mapping 设置与常见参数

控制当前字段是否被索引

null_value

copy_to

专题目录

Anaiysis 与 Analyzer

Analyzer 的组成

Elasticsearch 的内置分词器

Standard Analyzer

Simple Analyzer

Whitespace Analyzer

Stop Analyzer

Keyword Analyzer

Pattern Analyzer

Language Analyzer

使用 _analyzer Api

专题目录

概述

实践自定义分词器

Character Filters

Demo char_filter

html_strip 去除html标签

mapping 字符串替换

pattern_replace 正则表达式

Tokenizer

Demo tokenizer

path_hierarchy 通过路劲切分

Token tokenizer Filter

Demo filter

whitespace 空格

自定义 analyzer

自定义同义词需求解决

动态设置同义词的方案

专题目录

概述

实践

测试一下

分词原理

写时分词

读时分词

解决问题

自定义分词器

专题目录

概述

版本

相关链接

安装

docker方式安装

下载方式安装