mirror of
https://github.com/apache/doris.git
synced 2026-11-04 07:31:17 +00:00
281 lines
9.2 KiB
Python
281 lines
9.2 KiB
Python
#!/bin/env python
|
||
# -*- coding: utf-8 -*-
|
||
# Licensed to the Apache Software Foundation (ASF) under one
|
||
# or more contributor license agreements. See the NOTICE file
|
||
# distributed with this work for additional information
|
||
# regarding copyright ownership. The ASF licenses this file
|
||
# to you under the Apache License, Version 2.0 (the
|
||
# "License"); you may not use this file except in compliance
|
||
# with the License. You may obtain a copy of the License at
|
||
#
|
||
# http://www.apache.org/licenses/LICENSE-2.0
|
||
#
|
||
# Unless required by applicable law or agreed to in writing,
|
||
# software distributed under the License is distributed on an
|
||
# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
||
# KIND, either express or implied. See the License for the
|
||
# specific language governing permissions and limitations
|
||
# under the License.
|
||
|
||
"""
|
||
一个Palo的Task，
|
||
用于测试两类任务的相互影响时，持续执行一类任务
|
||
Date: 2015/01/22 10:49:31
|
||
"""
|
||
import random
|
||
import time
|
||
import threading
|
||
from lib import palo_client
|
||
|
||
|
||
class PaloTask(object):
|
||
"""
|
||
所有Task的父类
|
||
"""
|
||
def __init__(self, client):
|
||
self.client = client
|
||
|
||
def do_task(self):
|
||
"""
|
||
执行task, 在子类中实现
|
||
"""
|
||
pass
|
||
|
||
def wait_task(self):
|
||
"""
|
||
等待task执行结束，用于异步任务的状态监控, 需在子类中实现
|
||
"""
|
||
pass
|
||
|
||
def clean(self):
|
||
"""清理task中的残留"""
|
||
pass
|
||
|
||
|
||
class TaskThread(threading.Thread):
|
||
"""
|
||
启动一个线程循环去执行task
|
||
"""
|
||
def __init__(self, task):
|
||
self._exit_event = threading.Event()
|
||
self.task = task
|
||
threading.Thread.__init__(self)
|
||
self.setDaemon(True)
|
||
|
||
def stop(self):
|
||
"""
|
||
设置结束标记，会结束所有对象线程
|
||
"""
|
||
self._exit_event.set()
|
||
|
||
def run(self):
|
||
"""
|
||
启动线程
|
||
"""
|
||
while not self._exit_event.is_set():
|
||
self.task.wait_task()
|
||
self.task.do_task()
|
||
|
||
|
||
class SelectTask(PaloTask):
|
||
"""
|
||
查询任务
|
||
"""
|
||
def __init__(self, host, port, sql, database_name=None, expected_file_path=None,
|
||
user="root", password="", charset="utf8", delay=None, interval=None):
|
||
self.client = palo_client.get_client(host, port, database_name=database_name, user=user,
|
||
password=password, charset=charset)
|
||
self.sql = sql
|
||
if database_name is not None:
|
||
self.client.use(database_name)
|
||
self.expected_file_path = expected_file_path
|
||
self.delay = delay
|
||
if interval is None:
|
||
self.interval = 1
|
||
else:
|
||
self.interval = interval
|
||
|
||
def do_task(self):
|
||
"""
|
||
发送查询
|
||
"""
|
||
result = None
|
||
if self.delay is not None:
|
||
try_time = 0
|
||
while try_time < self.delay:
|
||
try:
|
||
result = self.client.execute(self.sql)
|
||
except:
|
||
try_time += self.interval
|
||
time.sleep(self.interval)
|
||
else:
|
||
break
|
||
else:
|
||
result = self.client.execute(self.sql)
|
||
|
||
if self.expected_file_path:
|
||
pass
|
||
return result
|
||
|
||
|
||
class BatchLoadTask(PaloTask):
|
||
"""
|
||
连续提交导入任务
|
||
"""
|
||
def __init__(self, host, port, database_name, load_label, load_data_list,
|
||
max_filter_ratio=None, timeout=None, is_wait=False, interval=None,
|
||
user="root", password="", charset="utf8", broker=None):
|
||
self.client = palo_client.get_client(host, port, database_name=database_name, user=user,
|
||
password=password, charset=charset)
|
||
self.client.use(database_name)
|
||
self.load_label = load_label
|
||
self.load_num = 0
|
||
self.load_data_list = load_data_list
|
||
self.max_filter_ratio = max_filter_ratio
|
||
self.timeout = timeout
|
||
self.is_wait = is_wait
|
||
self.broker = broker
|
||
if interval is None:
|
||
self.interval = 0
|
||
else:
|
||
self.interval = interval
|
||
|
||
def do_task(self):
|
||
"""
|
||
做导入任务
|
||
"""
|
||
load_label = "%s_%d" % (self.load_label, self.load_num)
|
||
ret = self.client.batch_load(load_label, self.load_data_list,
|
||
max_filter_ratio=self.max_filter_ratio,
|
||
timeout=self.timeout, is_wait=self.is_wait,
|
||
broker=self.broker)
|
||
assert ret
|
||
self.load_num += 1
|
||
time.sleep(self.interval)
|
||
|
||
|
||
class BulkLoadTask(PaloTask):
|
||
"""
|
||
连续提交小批量导入任务
|
||
"""
|
||
def __init__(self, host, port, be_host, webserver_port, database_name, table_family_name,
|
||
load_label, data_file, max_filter_ratio=None, timeout=None, is_wait=False,
|
||
user="root", password="", be_user="root", be_password="", charset="utf8", interval=0):
|
||
self.client = palo_client.get_client(host, port, database_name=database_name, user=user,
|
||
password=password, charset=charset)
|
||
self.be_host = be_host
|
||
self.webserver_port = webserver_port
|
||
self.database_name = database_name
|
||
self.table_family_name = table_family_name
|
||
self.load_label = load_label
|
||
self.load_num = 1
|
||
self.data_file = data_file
|
||
self.max_filter_ratio = max_filter_ratio
|
||
self.timeout = timeout
|
||
self.is_wait = is_wait
|
||
self.be_user = be_user
|
||
self.be_password = be_password
|
||
self.interval = interval
|
||
|
||
def do_task(self):
|
||
"""
|
||
做小批量导入任务
|
||
"""
|
||
load_label = "%s_%d" % (self.load_label, self.load_num)
|
||
ret = self.client.bulk_load(self.table_family_name, load_label, self.data_file,
|
||
self.max_filter_ratio, self.timeout, self.database_name, self.be_host,
|
||
self.webserver_port, self.is_wait, user=self.be_user, password=self.be_password)
|
||
assert ret
|
||
self.load_num += 1
|
||
time.sleep(self.interval)
|
||
|
||
|
||
class RollupTask(PaloTask):
|
||
"""
|
||
连续提交
|
||
"""
|
||
def __init__(self, host, port, database_name, table_family_name, rollup_table_name,
|
||
rollup_column_name_list, user="root", password="", charset="utf8", **kwargs):
|
||
self.client = palo_client.get_client(host, port, database_name=database_name, user=user,
|
||
password=password, charset=charset)
|
||
self.client.use(database_name)
|
||
self.table_family_name = table_family_name
|
||
self.rollup_table_name = rollup_table_name
|
||
self.rollup_num = 1
|
||
self.rollup_column_name_list = rollup_column_name_list
|
||
self.kwargs = kwargs
|
||
|
||
def do_task(self):
|
||
"""
|
||
做rollup
|
||
"""
|
||
rollup_table_name = "%s_%d" % (self.rollup_table_name, self.rollup_num)
|
||
self.rollup_num += 1
|
||
self.client.create_rollup_table(self.table_family_name, rollup_table_name,
|
||
self.rollup_column_name_list, is_wait=True)
|
||
|
||
|
||
class DeleteTask(PaloTask):
|
||
"""
|
||
数据删除任务, 循环使用delete_conditions_list中的删除条件，向palo发送数据删除命令
|
||
"""
|
||
def __init__(self, host, port, database_name, table_family_name, delete_conditions_list,
|
||
user="root", password="", charset="utf8", **kwargs):
|
||
"""
|
||
Parameters：
|
||
delete_conditions_list：由delete_condition_list(见PaloClient.delete)组成的list
|
||
"""
|
||
self.client = palo_client.get_client(host, port, database_name=database_name, user=user,
|
||
password=password, charset=charset)
|
||
self.client.use(database_name)
|
||
self.table_family_name = table_family_name
|
||
self.delete_conditions_list = delete_conditions_list
|
||
self.delete_conditions_index = 0
|
||
self.kwargs = kwargs
|
||
|
||
def do_task(self):
|
||
"""
|
||
执行一次数据删除，delete_conditions_index递增
|
||
"""
|
||
delete_condition_list = self.delete_conditions_list[self.delete_conditions_index % \
|
||
len(self.delete_conditions_list)]
|
||
self.delete_conditions_index += 1
|
||
try:
|
||
ret = self.client.delete(self.table_family_name, delete_condition_list, **self.kwargs)
|
||
except palo_client.PaloClientException as error:
|
||
print(str(error))
|
||
#TODO
|
||
pass
|
||
|
||
|
||
class SyncTask(PaloTask):
|
||
"""
|
||
执行同步任务，如查询，同步的insert、delete、update等
|
||
todo: 每个任务的参数相同，每个任务的结果不校验
|
||
"""
|
||
def __init__(self, func, *args, **kwargs):
|
||
self.func = func
|
||
self.args = args
|
||
self.kwargs = kwargs
|
||
self.succ_count = 0
|
||
self.error_count = 0
|
||
self.interval = None
|
||
|
||
def do_task(self):
|
||
"""
|
||
执行一次任务
|
||
"""
|
||
try:
|
||
self.func(*self.args, **self.kwargs)
|
||
self.succ_count += 1
|
||
except Exception as e:
|
||
self.error_count += 1
|
||
print(str(e))
|
||
if self.interval is None:
|
||
time.sleep(random.randint(0, 10) / 10.0)
|
||
else:
|
||
time.sleep(self.interval)
|
||
|
||
|
||
|