Tablestore SDK for Python kullanarak bir arama dizininde eşleşen verileri paralel olarak tarayın ve sırasız tam bir sonuç kümesini dışa aktarın.
Önkoşullar
Tablestore SDK for Python yükleyin ve bir istemci başlatın.
Açıklama
Parallel scan, bir arama dizinindeki sorguyla eşleşen tüm satırları dışa aktarır. Sonuçlar genel olarak sıralı değildir; sıralama ve toplama işlemleri desteklenmez. Sonuçları sıralamak veya toplamak ya da arama sonuçlarını son kullanıcılara döndürmek için Search API kullanın. Tek bir çalışan yapılandırmayı basitleştirirken, birden çok çalışan birden fazla bölümü eşzamanlı olarak okur ve genellikle daha yüksek aktarım hızı sağlar.
İş akışı şöyledir: Maksimum eşzamanlılık değeri olan splits_size ve oturum tanımlayıcısı session_id değerlerini almak için compute_splits yöntemini çağırın. Her çalışanı aynı sorgu, session_id ve max_parallel değerleriyle ve benzersiz bir current_parallel_id değeriyle yapılandırın. Her çalışan parallel_scan yöntemini çağırır ve sonraki sayfaları okumak için next_token kullanır. Son olarak, tüm çalışanların tamamlanmasını bekleyin ve sırasız sonuçlarını birleştirin.
Aynı session_id altındaki çalışanlar, ilk tarama başladığında bir veri anlık görüntüsü oluşturur. Dinamik şema güncellemeleri, yük devretme veya yük dengeleme gibi durumlar bir oturumu erken sonlandırabilir ve OTSSessionExpired hatası döndürebilir; ağ arızaları da taramayı kesintiye uğratabilir. Bu durumlarda, tamamlanmamış sonuçları atın, compute_splits yöntemini yeniden çağırın ve tüm çalışanları baştan başlatın. Tek bir arama dizininde eşzamanlı olarak en fazla 10 paralel tarama işi çalıştırılabilir.
Aşağıdaki örnekte bölümler hesaplanır ve ardından tek bir çalışan kullanılarak tüm veriler taranır.
splits = client.compute_splits("example_table", "example_index")
next_token = None
rows = []
while True:
scan_query = ScanQuery(
MatchAllQuery(),
limit=2000,
next_token=next_token,
current_parallel_id=0,
max_parallel=1,
alive_time=60,
)
response = client.parallel_scan(
"example_table",
"example_index",
scan_query,
splits.session_id,
ColumnsToGet(return_type=ColumnReturnType.ALL_FROM_INDEX),
)
rows.extend(response.rows)
next_token = response.next_token
if not next_token:
break
print(len(rows))Parametreler
Bölümleri hesaplama
compute_splits(table_name, index_name) aşağıdaki parametreleri içerir.
|
|
|
|
|
|
|
|
|
|
|
|
Tarama isteği
parallel_scan aşağıdaki parametreleri içerir.
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
Tarama yapılandırması
scan_query ScanQuery türündedir ve aşağıdaki parametreleri içerir.
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
Döndürülen sütunlar
columns_to_get ColumnsToGet türündedir ve aşağıdaki parametreleri içerir.
|
|
|
|
|
|
|
|
|
|
|
|
Yanıt
Bölüm bilgileri
compute_splits bölüm bilgilerini döndürür.
|
|
|
|
|
|
|
|
|
|
|
|
Tarama sonucu
parallel_scan bir tarama sonucu döndürür.
|
|
|
|
|
|
|
|
|
|
|
|
Tuple uyumlu yanıtlar
Parallel scan, Tablestore SDK for Python 5.2.0 sürümünden itibaren desteklenir ve yanıt nesneleri döndürür. 5.2.1 ve sonraki sürümlerde, tuple'ları almak için ComputeSplitsResponse.v1_response() ve ParallelScanResponse.v1_response() yöntemlerini çağırabilirsiniz. Yeni kodlarda yanıt nesnesi niteliklerine doğrudan erişin.
session_id, splits_size = splits.v1_response()
rows, next_token = response.v1_response()Örnekler
Birden çok çalışanla tarama
Aşağıdaki örnek, splits_size değerine dayalı olarak çalışanlar oluşturur. İş parçacığı havuzu istemci CPU çekirdek sayısını aşmaz ve tüm çalışanlar oturumu ve maksimum eşzamanlılığı paylaşır.
from concurrent.futures import ThreadPoolExecutor
import os
def scan_split(parallel_id, max_parallel, session_id):
rows = []
next_token = None
while True:
scan_query = ScanQuery(
MatchAllQuery(),
limit=2000,
next_token=next_token,
current_parallel_id=parallel_id,
max_parallel=max_parallel,
alive_time=60,
)
response = client.parallel_scan(
"example_table",
"example_index",
scan_query,
session_id,
ColumnsToGet(return_type=ColumnReturnType.ALL_FROM_INDEX),
)
rows.extend(response.rows)
next_token = response.next_token
if not next_token:
return rows
splits = client.compute_splits("example_table", "example_index")
worker_count = min(splits.splits_size, os.cpu_count() or 1)
with ThreadPoolExecutor(max_workers=worker_count) as executor:
futures = [
executor.submit(
scan_split,
parallel_id,
splits.splits_size,
splits.session_id,
)
for parallel_id in range(splits.splits_size)
]
all_rows = [row for future in futures for row in future.result()]
print(len(all_rows))