从ElasticSearch迁移至openGauss
- 环境准备
- 迁移操作

从ElasticSearch迁移至openGauss

本教程通过Python编程语言，支持将ElasticSearch数据迁移至openGauss实例中

环境准备

已部署7.3 及以上版本的ElasticSearch实例
已部署7.0.0-RC1 及以上版本的openGauss实例
已安装3.8 及以上版本的Python环境
已安装涉及的Python库

1. pip3 install psycopg2

迁移操作

参考如下迁移脚本elasticSearchToOpenGauss.py，根据本地部署的elasticsearch与openGauss进行配置修改：

elasticsearchToOpenGauss.py迁移脚本如下：

1. import requests
2. import psycopg2
3. import json
4. import re
5. import requests
6. from typing import List, Dict, Any, Optional, Union

8. # Elasticsearch 配置信息
9. es_url = 'http://ip:port'  # Elasticsearch 服务器地址
10. es_index = 'your_es_index'  # Elasticsearch 索引名

12. # openGauss 配置信息
13. db_host = '127.0.0.1'
14. db_port = 5432
15. db_name = 'your_opengauss_db'
16. db_user = 'user_name'
17. db_password = 'xxxxxx'

19. RESERVED_KEYWORDS = {
20. "select", "insert", "update", "delete", "drop", "table", "from", "where", "group",
21. "by", "having", "order", "limit", "join", "inner", "left", "right", "full", "union",
22. "all", "distinct", "as", "on", "and", "or", "not", "null", "true", "false", "case",
23. "when", "then", "else", "end", "exists", "like", "in", "between", "is", "like",
24. "references", "foreign", "primary", "key", "unique", "check", "default", "constraint",
25. "index", "unique", "varchar", "text", "int", "bigint", "smallint", "boolean", "timestamp"
26. }
27. # 从 Elasticsearch 获取数据
28. def fetch_data_from_es():
29. query = {
30. "query": {
31. "match_all": {}
32. },
33. "_source": True  # 获取所有字段
34. }
35. response = requests.get(f'{es_url}/{es_index}/_search', json=query)
36. if response.status_code == 200:
37. return response.json()['hits']['hits']
38. else:
39. raise Exception(f"Failed to fetch data from Elasticsearch: {response.status_code}, {response.text}")

42. # 获取索引映射信息
43. def fetch_mapping(es_url, es_index):
44. response = requests.get(f'{es_url}/{es_index}/_mapping')
45. if response.status_code == 200:
46. return response.json()
47. else:
48. raise Exception(f"Failed to fetch mapping: {response.status_code}, {response.text}")

51. def get_field_type(es_url: str, es_index: str, field_name: str) -> str:
52. """ 获取 Elasticsearch 字段的类型 """
53. mappings = fetch_mapping(es_url, es_index)
54. print(f"Field name: {field_name}")
55. print(f"map: {mappings}")
56. # 获取 properties 字段
57. properties = mappings.get(es_index, {}).get('mappings', {}).get('properties', {})
58. # 遍历并查找字段的类型
59. field_type = 'text'  # 默认类型为 'text'
60. if field_name in properties:
61. field_type = properties[field_name].get('type', 'text')
62. elif 'fields' in properties.get(field_name, {}):
63. # 如果字段有子字段（比如 keyword），获取 'keyword' 类型
64. field_type = properties[field_name]['fields'].get('keyword', {}).get('type', 'text')
65. return field_type

67. def convert_dict_to_jsonb(value):
68. # 如果 value 是字典类型，递归调用该函数处理其中的每个元素
69. if isinstance(value, dict):
70. return json.dumps({k: convert_dict_to_jsonb(v) for k, v in value.items()})
71. # 如果 value 是列表类型，递归处理其中的每个元素
72. elif isinstance(value, list):
73. return json.dumps([convert_dict_to_jsonb(v) for v in value])
74. # 如果是其他类型（如字符串、数字），直接返回该值
75. else:
76. return value

78. # 映射 Elasticsearch 数据类型到 openGauss 类型
79. def map_to_opengauss_type(es_type: str, dim: Optional[int] = None) -> str:
80. """Map Elasticsearch types to openGauss types"""
81. if isinstance(es_type, (dict, list)):  # 如果 es_type 是字典类型，则需要特殊处理
82. return 'JSONB'
83. type_map = {
84. "long": "BIGINT",  # 大整数
85. "integer": "INTEGER",  # 整数
86. "short": "SMALLINT",  # 小整数
87. "byte": "SMALLINT",  # 小字节
88. "float": "REAL",  # 浮点数
89. "double": "DOUBLE PRECISION",  # 双精度浮点数
90. "boolean": "BOOLEAN",  # 布尔值
91. "keyword": "VARCHAR",  # 关键字（字符串类型）
92. "text": "TEXT",  # 长文本
93. "date": "TIMESTAMP",  # 日期类型
94. "binary": "BYTEA",  # 二进制数据
95. "geo_point": "POINT",  # 地理坐标（经纬度）
96. "geo_shape": "GEOMETRY",  # 复杂地理形状
97. "nested": "JSONB",  # 嵌套对象
98. "object": "JSONB",  # 对象
99. "ip": "INET",  # IP 地址
100. "scaled_float": "REAL",  # 扩展浮动类型（带缩放的浮动）
101. "float_vector": f"VECTOR({dim})" if dim else "VECTOR",  # 浮动向量类型
102. "dense_vector": f"VECTOR({dim})" if dim else "VECTOR",  # 稠密向量类型
103. "binary_vector": f"BIT({dim})" if dim else "BIT",  # 二进制向量类型
104. "half_float": "REAL",  # 半精度浮动
105. "unsigned_long": "BIGINT",  # 无符号长整数
106. "date_nanos": "TIMESTAMP",  # 高精度日期时间
107. "alias": "TEXT",  # 别名（通常是字段的别名）
108. }

110. # 如果 es_type 在映射表中，直接返回映射后的类型
111. if es_type in type_map:
112. print(f"es_type:{es_type} ----- og_type: {type_map[es_type]}")
113. return type_map[es_type]
114. else:
115. print(f"Warning: Unsupported Elasticsearch type '{es_type}', defaulting to 'TEXT'")
116. return 'TEXT'  # 默认使用 TEXT 类型

119. # 函数：将非法字符替换为下划线
120. def sanitize_name(field_name: str) -> str:
121. """处理字段名，确保不会与保留字冲突，且将非字母数字字符替换为下划线"""
122. # 将所有非字母数字字符替换为下划线
123. sanitized_name = re.sub(r'[^a-zA-Z0-9_]', '_', field_name)

125. # 如果是保留字，则加双引号
126. if sanitized_name.lower() in RESERVED_KEYWORDS:
127. return f'"{sanitized_name}"'

129. return sanitized_name

132. # 创建 openGauss 表
133. def create_table_in_opengauss(es_url, es_index, table_name):
134. columns_definition = ['id VARCHAR PRIMARY KEY']  # 增加 id 主键字段
135. seen_fields = set()  # 用于记录已经处理过的字段名

137. # 获取 properties 字段
138. properties = fetch_mapping(es_url, es_index).get(es_index, {}).get('mappings', {}).get('properties', {})

140. # 遍历每个字段
141. for field, field_info in properties.items():
142. # 如果该字段已经处理过，跳过
143. if field in seen_fields:
144. continue
145. # 获取字段的类型
146. es_type = field_info.get('type', 'text')
147. dim = field_info.get('dims', 0) if isinstance(field_info, dict) else 0
148. field_type = map_to_opengauss_type(es_type, dim)
149. sanitized_field_name = sanitize_name(field)
150. seen_fields.add(field)
151. columns_definition.append(f"{sanitized_field_name} {field_type}")
152. # 生成表创建 SQL
153. columns_str = ", ".join(columns_definition)
154. create_table_sql = f"DROP TABLE IF EXISTS {sanitize_name(table_name)}; CREATE TABLE {sanitize_name(table_name)} ({columns_str});"
155. try:
156. # 建立数据库连接并执行创建表 SQL
157. connection = psycopg2.connect(
158. host=db_host,
159. port=db_port,
160. dbname=db_name,
161. user=db_user,
162. password=db_password
163. )
164. cursor = connection.cursor()
165. cursor.execute(create_table_sql)
166. connection.commit()
167. print(f"Table {sanitize_name(table_name)} created successfully.")
168. except Exception as e:
169. print(f"Error while creating table {sanitize_name(table_name)}: {e}")
170. finally:
171. if connection:
172. cursor.close()
173. connection.close()

175. # 将数据插入到 openGauss 表中
176. def insert_data_to_opengauss(table_name, es_source, es_id):
177. try:
178. # 建立数据库连接
179. connection = psycopg2.connect(
180. host=db_host,
181. port=db_port,
182. dbname=db_name,
183. user=db_user,
184. password=db_password
185. )
186. cursor = connection.cursor()

188. # 动态生成插入 SQL 语句
189. sanitized_columns = ['id'] + [sanitize_name(col) for col in es_source.keys()]  # 清理列名
190. values = [es_id]

192. # 处理每一列的数据类型，必要时进行转换
193. for column in es_source:
194. value = es_source[column]
195. if isinstance(value, (dict, list)):
196. # 如果是字典类型，转换为 JSONB
197. value = convert_dict_to_jsonb(value)
198. values.append(value)

200. columns_str = ', '.join(sanitized_columns)
201. values_str = ', '.join(['%s'] * len(values))

203. insert_sql = f"INSERT INTO {sanitize_name(table_name)} ({columns_str}) VALUES ({values_str})"
204. cursor.execute(insert_sql, values)

206. # 提交事务
207. connection.commit()

209. except Exception as e:
210. print(f"Error while inserting data into {table_name}: {e}")
211. finally:
212. if connection:
213. cursor.close()
214. connection.close()

216. # 主函数
217. def main():
218. try:
219. es_data = fetch_data_from_es()
220. table_name = es_index  # 可以使用索引名作为表名

222. create_table_in_opengauss(es_url, es_index, table_name)
223. for record in es_data:
224. es_source = record['_source']  # 获取 Elasticsearch 文档中的数据
225. es_id = record['_id']
226. insert_data_to_opengauss(table_name, es_source, es_id)
227. print(f"Successfully inserted data into table {table_name}.")

230. except Exception as e:
231. print(f"Migration failed: {e}")

234. if __name__ == "__main__":
235. main()

运行迁移脚本，并查看输出结果：

1. python3 elasticSearchToOpenGauss.py

登录openGauss，查看数据是否完成迁移：

4.1 登录omm超级用户：

1. $ su - omm 2. $ gsql -d postgres -p 5432

4.2 查看迁移表数据量：

1. $ \c your_opengauss_db 2. $ \d #查看迁移后的表名table_name 3. $ select count(*) from table_name;

从ElasticSearch迁移至openGauss DataVec

从ElasticSearch迁移至openGauss

环境准备

迁移操作