YugabyteDB에서 집계 병렬화

이전 게시물에서 재귀 CTE 및 커버링 범위 인덱스를 사용하여 느슨한 인덱스 스캔을 시뮬레이트하는 방법을 보았습니다. 각각의 마지막 값truck_id을 얻기 위해 사용했습니다. 이제 각truck_id당 판독값을 계산하고 싶습니다. 이것은 모든 행을 스캔해야 합니다. 그러나 이전 기술을 사용하여 이 작업을 병렬로 수행하는 명령문을 생성할 수 있습니다.

예시

나는 2000대의 트럭 각각에 대해 10000개의 판독값을 삽입한 위치에서 생성된 truck_readings 테이블과 truck_last_reading 인덱스를 사용하고 있습니다.

10개의 태블릿으로 인덱스를 생성했다는 점에 유의하십시오(실제로 이것은 테이블이 커질 때 auto_splitting으로 수행됨).

create index truck_last_reading on truck_readings 
( truck_id asc, ts asc) 
split at values((100),(200),(300),(400),(500),(600),(700),(800),(1000));

방법 1: 병렬 문 생성

먼저 재귀 CTE 기술을 사용하여 각각truck_id을 나열하고 개수를 입력할 자리 표시자가 있는 테이블을 만듭니다.

create table tmp_partial_count as
with recursive truck_last_reading as (
 (
  select
   last_truck_last_reading.truck_id
   from truck_readings last_truck_last_reading
  order by
   last_truck_last_reading.truck_id desc,
   last_truck_last_reading.ts desc
  limit 1
 )
union all
 select
  next_truck_last_reading.truck_id
 from truck_last_reading, lateral
 (
  select truck_id from truck_readings
  where truck_id < truck_last_reading.truck_id
  order by truck_id desc, ts desc limit 1
 )
 as next_truck_last_reading
) select truck_id, null::int as partial_count
from truck_last_reading
;

여기서는 2초가 걸립니다.

...
SELECT 2000
Time: 2464.776 ms (00:02.465)
yugabyte=#

그런 다음 이 테이블에서 각각의 판독값을 계산하는 명령문truck_id을 생성하고 임시 테이블에서 업데이트합니다. psql 와 함께 백그라운드에서 실행할 & 문을 생성합니다. 연결 수를 제한하기 위해 작업 50개마다 wait를 추가합니다.

\pset format unaligned
\pset tuples_only on
\pset footer off
\o tmp_partial_count.sh
select format($$
psql -d postgres://yugabyte:YugabyteDB@yb0.pachot.net:5433/yugabyte -c 'update tmp_partial_count 
set partial_count=(
select count(*) from truck_readings where truck_id=%s
) where truck_id=%s' &
%s
$$
,truck_id,truck_id
,case when mod(row_number()over(), 200 )=0 then 'wait' end
)
from tmp_partial_count where partial_count is null
;
select 'wait';
\o

이렇게 하면 tmp_partial_count.sh에 스크립트가 생성됩니다.

...
yugabyte-# from tmp_partial_count where partial_count is null
yugabyte-# ;
select 'wait';
\oTime: 47.225 ms
yugabyte=# select 'time wait';
Time: 11.529 ms
yugabyte=# \o

부분 카운트가 없는 행에 대해서만 생성합니다.

나는 이것을 실행할 수 있습니다 :

\! time sh tmp_partial_count.sh

이것은 내 연구실(소규모 연구실이지만 확장할 수 있다는 점)에서 30분 동안 지속되었습니다.

...
UPDATE 1
UPDATE 1
UPDATE 1

real    0m55.229s
user    0m5.690s
sys     0m4.904s
yugabyte=#

내 인덱스가 Index Only Scan과 함께 사용되는지 확인할 수 있습니다. 이것이 확장되는 이유입니다(각 병렬 쿼리는 다른 범위를 읽습니다).

yugabyte=# explain (analyze, costs off) 
           update tmp_partial_count set partial_count=(
             select count(*) from truck_readings
             where truck_id=42
           ) where truck_id=42;

                                                       QUERY PLAN
-------------------------------------------------------------------------------------------------------------------------
 Update on tmp_partial_count (actual time=50.422..50.422 rows=0 loops=1)
   InitPlan 1 (returns $0)
     ->  Aggregate (actual time=41.160..41.160 rows=1 loops=1)
           ->  Index Only Scan using truck_last_reading on truck_readings (actual time=4.255..40.626 rows=10000 loops=1)
                 Index Cond: (truck_id = 42)
                 Heap Fetches: 0
   ->  Seq Scan on tmp_partial_count (actual time=50.300..50.394 rows=1 loops=1)
         Filter: (truck_id = 42)
         Rows Removed by Filter: 1999
 Planning Time: 0.091 ms
 Execution Time: 53.597 ms
 Peak Memory Usage: 24 kB
(12 rows)

다시 말하지만 제 연구실은 이곳이 작습니다. 대규모 클러스터의 경우 인덱스가 여러 태블릿에서 범위별로 분할될 수 있습니다. 활성화한 경우 자동으로 수행됩니다auto-split.

마지막으로 모든 부분 업데이트가 있는지 확인합니다.

yugabyte=# 
           select count(*),count(partial_count),sum(partial_count)
           from tmp_partial_count;

 count | count |   sum
-------+-------+----------
  2000 |  2000 | 20000000
(1 row)

방법 2: Seq 스캔 병렬 처리 사용

저는 YugabyteDB 2.15로 이 글을 쓰고 있으며 향후 버전에서 집계를 병렬화하고 푸시다운하는 일부 최적화를 기대할 수 있습니다. 먼저 하나의 쿼리에 대한 실행 계획을 확인하겠습니다.

yugabyte=#
             explain (costs off, analyze)
             select   truck_id, count(*) from truck_readings 
             where truck_id is not null
             group by truck_id
;

                                      QUERY PLAN
---------------------------------------------------------------------------------------
 HashAggregate (actual time=65548.176..65548.601 rows=2000 loops=1)
   Group Key: truck_id
   ->  Seq Scan on truck_readings (actual time=3.170..62186.109 rows=20000000 loops=1)
 Planning Time: 3.297 ms
 Execution Time: 65549.386 ms
 Peak Memory Usage: 12665 kB
(6 rows)

1분 정도 걸립니다. 내 실험실에서 이것은 위의 복잡한 방법보다 느리지 않지만 더 큰 볼륨에서는 한 번에 하나의 태블릿을 읽기 때문에 확장되지 않습니다.

YugabyteDB는 태블릿을 병렬로 읽기 위해 Seq 스캔을 병렬화할 수 있습니다(기본값--ysql_select_parallelism=-1은 tserver의 수에서 계산됩니다: 서버당 2, 2와 16 사이의 경계](https://github.com/yugabyte/yugabyte-db/blob/2.15.2/src/yb/yql/pggate/pg_doc_op.cc#L762)). 제 연구실에는 3개의 tserver가 있습니다. 모든 행을 병렬로 읽으면 단일 YSQL 프로세스(PostgreSQL 백엔드)가 포화되기 때문에 YugabyteDB는 푸시다운(Remote Filter ) 표현식이 있을 때만 SeqScan을 병렬화했습니다.

yugabyte=#   set yb_enable_expression_pushdown=true;
yugabyte=#
             explain (costs off, analyze) 
             select   truck_id, count(*) from truck_readings 
             where truck_id is not null
             group by truck_id
;
                                      QUERY PLAN
---------------------------------------------------------------------------------------
 HashAggregate (actual time=39623.390..39623.786 rows=2000 loops=1)
   Group Key: truck_id
   ->  Seq Scan on truck_readings (actual time=4.386..34862.379 rows=20000000 loops=1)
         Remote Filter: (truck_id IS NOT NULL)
 Planning Time: 0.071 ms
 Execution Time: 39624.001 ms
 Peak Memory Usage: 417 kB
(7 rows)

Time: 39636.634 ms (00:39.637)

여기서는 훨씬 간단한 방법으로 더 나은 결과를 얻습니다. 그러나 이것은 서버와 데이터에 따라 다르다는 것을 기억하십시오. 태블릿 읽기는 병렬화되었지만 단일 백엔드에서 모든 행을 가져와 집계해야 합니다. 즉, 여러 연결에서 병렬로 쿼리를 실행하는 것만큼 확장성이 떨어집니다.