258 lines
12 KiB
PHP
258 lines
12 KiB
PHP
<?php
|
||
|
||
namespace App\Console\Commands;
|
||
|
||
use GuzzleHttp\Psr7\StreamWrapper;
|
||
use Illuminate\Console\Command;
|
||
use Illuminate\Support\Facades\Schema;
|
||
use Illuminate\Support\Facades\Storage;
|
||
use Illuminate\Support\Str;
|
||
use Modules\EcommerceFinance\Repositories\ColumnarBusinessMirrorRepository;
|
||
use Modules\EcommerceFinance\Contracts\BusinessMirrorDataset;
|
||
use Modules\EcommerceFinance\Services\BusinessDataPlacement;
|
||
use Modules\EcommerceFinance\Services\BusinessMirrorDatasetRegistry;
|
||
use Modules\EcommerceFinance\Services\ClickHouseClient;
|
||
use Modules\EcommerceFinance\Services\ClickHouseSchemaManager;
|
||
use RuntimeException;
|
||
use Throwable;
|
||
|
||
class BackfillBusinessDataColumnar extends Command
|
||
{
|
||
protected $signature = 'business-data:backfill-columnar
|
||
{dataset : config/upstream_mirrors.php 中的数据集键}
|
||
{--batch=5000 : 每批从现有 MariaDB 镜像读取的行数}
|
||
{--max-batches=10 : 本轮最多批次数，0 表示直到完成}
|
||
{--restart : 使用新代次从头重建，不删除旧代次}
|
||
{--archive : 对账通过后将 32 个哈希分区分别写入 Parquet 归档}';
|
||
protected $description = '从现有 MariaDB 镜像断点复制到 ClickHouse，不访问上游、不删除原表';
|
||
|
||
public function handle(
|
||
BusinessMirrorDatasetRegistry $datasets,
|
||
ClickHouseSchemaManager $clickHouseSchemas,
|
||
ColumnarBusinessMirrorRepository $columnar,
|
||
BusinessDataPlacement $placements,
|
||
ClickHouseClient $clickHouse,
|
||
): int {
|
||
$datasetKey = (string) $this->argument('dataset');
|
||
|
||
try {
|
||
$dataset = $datasets->get($datasetKey);
|
||
$targetStore = $placements->targetStore($datasetKey);
|
||
if (! in_array($targetStore, ['clickhouse', 'parquet'], true)) {
|
||
throw new RuntimeException('business_data_dataset_not_columnar');
|
||
}
|
||
if (! $clickHouse->enabled() || ! $clickHouse->ping()) {
|
||
throw new RuntimeException('business_clickhouse_unavailable');
|
||
}
|
||
|
||
$connection = $placements->connection();
|
||
$schema = Schema::connection($connection->getName());
|
||
if (! $schema->hasTable('business_data_placements') || ! $schema->hasTable($dataset->targetTable())) {
|
||
throw new RuntimeException('business_data_source_or_state_table_missing');
|
||
}
|
||
|
||
$columns = $clickHouseSchemas->ensure($dataset);
|
||
$columnNames = array_map(static fn (array $column): string => (string) $column['column_name'], $columns);
|
||
$state = $placements->state($datasetKey);
|
||
$restart = (bool) $this->option('restart');
|
||
$verified = ! $restart
|
||
&& is_array($state)
|
||
&& in_array((string) ($state['phase'] ?? ''), ['verified', 'active'], true)
|
||
&& ! empty($state['backfill_generation'])
|
||
&& ! empty($state['verified_at']);
|
||
if ($verified) {
|
||
$generation = (string) $state['backfill_generation'];
|
||
$this->info(sprintf(
|
||
'%s：复用已对账代次 %s（%d 行）；MariaDB 原表仍保留。',
|
||
$datasetKey,
|
||
$generation,
|
||
(int) ($state['target_rows'] ?? 0),
|
||
));
|
||
if ((bool) $this->option('archive')) {
|
||
$this->archive($datasetKey, $dataset, $generation, $columnar, $placements);
|
||
}
|
||
|
||
return self::SUCCESS;
|
||
}
|
||
$resume = ! $restart
|
||
&& is_array($state)
|
||
&& in_array((string) ($state['phase'] ?? ''), ['backfilling', 'failed'], true)
|
||
&& ! empty($state['backfill_generation']);
|
||
|
||
$generation = $resume ? (string) $state['backfill_generation'] : (string) Str::uuid();
|
||
$cursor = $resume ? (string) ($state['backfill_cursor'] ?? '') : '';
|
||
$cutoff = $resume && ! empty($state['source_cutoff_at'])
|
||
? (string) $state['source_cutoff_at']
|
||
: now()->format('Y-m-d H:i:s.u');
|
||
$sourceQuery = $connection->table($dataset->targetTable())
|
||
->where('_mirror_synced_at', '<=', $cutoff);
|
||
$sourceRows = $resume ? (int) $state['source_rows'] : (int) (clone $sourceQuery)->count();
|
||
$backfilledRows = $resume ? (int) $state['backfilled_rows'] : 0;
|
||
|
||
$placements->save($datasetKey, [
|
||
'target_store' => $targetStore,
|
||
'phase' => 'backfilling',
|
||
'backfill_generation' => $generation,
|
||
'backfill_cursor' => $cursor === '' ? null : $cursor,
|
||
'source_cutoff_at' => $cutoff,
|
||
'source_rows' => $sourceRows,
|
||
'backfilled_rows' => $backfilledRows,
|
||
'target_rows' => 0,
|
||
'verified_at' => null,
|
||
'last_error_code' => null,
|
||
]);
|
||
|
||
$batchSize = min(max((int) $this->option('batch'), 100), 10000);
|
||
$maxBatches = max((int) $this->option('max-batches'), 0);
|
||
$batches = 0;
|
||
|
||
while ($maxBatches === 0 || $batches < $maxBatches) {
|
||
$query = $connection->table($dataset->targetTable())
|
||
->where('_mirror_synced_at', '<=', $cutoff)
|
||
->orderBy('_mirror_key')
|
||
->limit($batchSize);
|
||
if ($cursor !== '') {
|
||
$query->where('_mirror_key', '>', $cursor);
|
||
}
|
||
$rows = $query->get(['_mirror_key', ...$columnNames, '_mirror_synced_at'])
|
||
->map(static fn (object $row): array => (array) $row)
|
||
->all();
|
||
if ($rows === []) {
|
||
break;
|
||
}
|
||
|
||
$columnar->storeBackfillBatch($dataset, $columns, $rows, $generation);
|
||
$cursor = (string) $rows[array_key_last($rows)]['_mirror_key'];
|
||
$backfilledRows += count($rows);
|
||
$batches++;
|
||
$placements->save($datasetKey, [
|
||
'phase' => 'backfilling',
|
||
'backfill_generation' => $generation,
|
||
'backfill_cursor' => $cursor,
|
||
'source_cutoff_at' => $cutoff,
|
||
'source_rows' => $sourceRows,
|
||
'backfilled_rows' => $backfilledRows,
|
||
'last_error_code' => null,
|
||
]);
|
||
|
||
if (count($rows) < $batchSize) {
|
||
break;
|
||
}
|
||
}
|
||
|
||
if ($backfilledRows < $sourceRows) {
|
||
$this->info(sprintf('%s：本轮 %d 批，已复制 %d/%d 行；可重复执行续传。', $datasetKey, $batches, $backfilledRows, $sourceRows));
|
||
|
||
return self::SUCCESS;
|
||
}
|
||
|
||
$targetRows = $columnar->count($dataset, $generation);
|
||
$distinctKeys = $columnar->distinctKeys($dataset, $generation);
|
||
if ($targetRows !== $sourceRows || $distinctKeys !== $sourceRows) {
|
||
$mismatch = sprintf(
|
||
'business_data_backfill_count_mismatch:%d:%d:%d',
|
||
$sourceRows,
|
||
$targetRows,
|
||
$distinctKeys,
|
||
);
|
||
$placements->save($datasetKey, [
|
||
'phase' => 'failed',
|
||
'target_rows' => $targetRows,
|
||
'last_error_code' => $mismatch,
|
||
]);
|
||
throw new RuntimeException($mismatch);
|
||
}
|
||
|
||
$placements->save($datasetKey, [
|
||
'phase' => 'verified',
|
||
'target_rows' => $targetRows,
|
||
'verified_at' => now(),
|
||
'last_error_code' => null,
|
||
]);
|
||
$this->info(sprintf('%s：MariaDB 与 ClickHouse 已对账，代次 %s 共 %d 行；尚未切换页面或删除原表。', $datasetKey, $generation, $targetRows));
|
||
|
||
if ((bool) $this->option('archive')) {
|
||
$this->archive($datasetKey, $dataset, $generation, $columnar, $placements);
|
||
}
|
||
|
||
return self::SUCCESS;
|
||
} catch (Throwable $exception) {
|
||
$errorCode = preg_match('/^(business_|clickhouse_)[A-Za-z0-9_:.-]+$/', $exception->getMessage()) === 1
|
||
? $exception->getMessage()
|
||
: $exception::class;
|
||
if (isset($placements) && Schema::connection($placements->connection()->getName())->hasTable('business_data_placements')) {
|
||
try {
|
||
$placements->save($datasetKey, [
|
||
'phase' => 'failed',
|
||
'last_error_code' => mb_substr($errorCode, 0, 120),
|
||
]);
|
||
} catch (Throwable) {
|
||
// Preserve the original failure when the control store is unavailable.
|
||
}
|
||
}
|
||
report($exception);
|
||
$this->error($datasetKey . '：列式回填失败（' . $errorCode . '），现有 MariaDB 数据未删除、页面未切换。');
|
||
|
||
return self::FAILURE;
|
||
}
|
||
}
|
||
|
||
private function archive(
|
||
string $datasetKey,
|
||
BusinessMirrorDataset $dataset,
|
||
string $generation,
|
||
ColumnarBusinessMirrorRepository $columnar,
|
||
BusinessDataPlacement $placements,
|
||
): void {
|
||
$diskName = (string) config('business_data.archive.disk', 'business_uploads_qiniu');
|
||
$prefix = trim((string) config('business_data.archive.prefix', 'business-data-archive'), '/');
|
||
$slug = trim((string) preg_replace('/[^A-Za-z0-9_.-]+/', '-', $datasetKey), '-');
|
||
$disk = Storage::disk($diskName);
|
||
|
||
for ($partition = 0; $partition < 32; $partition++) {
|
||
$partitionKey = sprintf('hash-%02d', $partition);
|
||
$objectKey = sprintf('%s/%s/%s/%s.parquet', $prefix, $slug, $generation, $partitionKey);
|
||
$existing = $placements->connection()->table('business_data_archives')
|
||
->where('dataset_key', $datasetKey)
|
||
->where('generation', $generation)
|
||
->where('partition_key', $partitionKey)
|
||
->first();
|
||
if ($existing !== null
|
||
&& (string) ($existing->status ?? '') === 'verified'
|
||
&& $disk->exists($objectKey)
|
||
&& (int) $disk->size($objectKey) === (int) ($existing->byte_size ?? 0)) {
|
||
continue;
|
||
}
|
||
$stream = $columnar->parquetPartition($dataset, $generation, $partition);
|
||
$resource = StreamWrapper::getResource($stream);
|
||
try {
|
||
$disk->writeStream($objectKey, $resource);
|
||
} finally {
|
||
fclose($resource);
|
||
}
|
||
|
||
$bytes = (int) $disk->size($objectKey);
|
||
$stats = $columnar->partitionStats($dataset, $generation, $partition);
|
||
$rowCount = $stats['row_count'];
|
||
if ($bytes <= 0 || ($rowCount > 0 && $bytes < 8)) {
|
||
throw new RuntimeException('business_data_archive_verification_failed');
|
||
}
|
||
$placements->connection()->table('business_data_archives')->updateOrInsert(
|
||
['dataset_key' => $datasetKey, 'generation' => $generation, 'partition_key' => $partitionKey],
|
||
[
|
||
'object_key' => $objectKey,
|
||
'row_count' => $rowCount,
|
||
'byte_size' => $bytes,
|
||
'logical_checksum' => $stats['logical_checksum'],
|
||
'status' => 'verified',
|
||
'verified_at' => now(),
|
||
'created_at' => now(),
|
||
'updated_at' => now(),
|
||
],
|
||
);
|
||
}
|
||
|
||
$this->info($datasetKey . '：32 个 Parquet 哈希分区已写入对象存储并登记清单；MariaDB 原表仍保留。');
|
||
}
|
||
}
|