无服务器权限下的数据迁移方案API驱动与ETL工具

class APIDataMigrator {
    constructor(sourceConfig, targetConfig) {
        this.sourceConfig = sourceConfig;
        this.targetConfig = targetConfig;
        this.batchSize = sourceConfig.batchSize || 100;
        this.concurrency = sourceConfig.concurrency || 5;
    }
    
    async authenticate() {
        const sourceToken = await this.getSourceToken();
        const targetToken = await this.getTargetToken();
        
        return { sourceToken, targetToken };
    }
    
    async getSourceToken() {
        const response = await fetch(`${this.sourceConfig.baseUrl}/auth/login`, {
            method: 'POST',
            headers: { 'Content-Type': 'application/json' },
            body: JSON.stringify({
                username: this.sourceConfig.username,
                password: this.sourceConfig.password
            })
        });
        
        const data = await response.json();
        return data.token;
    }
    
    async getTargetToken() {
        const response = await fetch(`${this.targetConfig.baseUrl}/oauth/token`, {
            method: 'POST',
            headers: { 'Content-Type': 'application/json' },
            body: JSON.stringify({
                client_id: this.targetConfig.clientId,
                client_secret: this.targetConfig.clientSecret,
                grant_type: 'client_credentials'
            })
        });
        
        const data = await response.json();
        return data.access_token;
    }
    
    async getAllData(endpoint) {
        const token = await this.getSourceToken();
        let allData = [];
        let page = 1;
        let hasMore = true;
        
        while (hasMore) {
            const response = await fetch(
                `${this.sourceConfig.baseUrl}${endpoint}?page=${page}&pageSize=${this.batchSize}`,
                { headers: { Authorization: `Bearer ${token}` } }
            );
            
            const data = await response.json();
            allData = allData.concat(data.items);
            
            hasMore = data.hasMore && data.items.length > 0;
            page++;
        }
        
        return allData;
    }
    
    async transformData(sourceData) {
        return sourceData.map(item => ({
            id: item.id,
            name: item.name,
            email: item.email,
            phone: item.phone,
            createdAt: item.createdAt,
            updatedAt: item.updatedAt
        }));
    }
    
    async validateData(data) {
        const errors = [];
        
        for (const item of data) {
            if (!item.name) {
                errors.push({ id: item.id, error: 'Name is required' });
            }
            if (!item.email) {
                errors.push({ id: item.id, error: 'Email is required' });
            }
        }
        
        return { valid: errors.length === 0, errors };
    }
    
    async batchWrite(targetToken, data) {
        const batches = this.chunkArray(data, this.batchSize);
        let successCount = 0;
        let failCount = 0;
        
        for (const batch of batches) {
            try {
                const response = await fetch(`${this.targetConfig.baseUrl}/api/batch`, {
                    method: 'POST',
                    headers: {
                        Authorization: `Bearer ${targetToken}`,
                        'Content-Type': 'application/json'
                    },
                    body: JSON.stringify(batch)
                });
                
                const result = await response.json();
                successCount += result.successCount;
                failCount += result.failCount;
            } catch (error) {
                failCount += batch.length;
            }
        }
        
        return { successCount, failCount };
    }
    
    async run(endpoint) {
        console.log('Starting API data migration...');
        
        console.log('Authenticating...');
        const tokens = await this.authenticate();
        console.log('Authentication successful');
        
        console.log('Fetching data...');
        const sourceData = await this.getAllData(endpoint);
        console.log(`Fetched ${sourceData.length} records`);
        
        console.log('Transforming data...');
        const transformedData = await this.transformData(sourceData);
        console.log('Data transformation complete');
        
        console.log('Validating data...');
        const validation = await this.validateData(transformedData);
        if (!validation.valid) {
            console.error(`Validation failed: ${validation.errors.length} errors`);
            return;
        }
        console.log('Data validation complete');
        
        console.log('Writing data...');
        const result = await this.batchWrite(tokens.targetToken, transformedData);
        console.log(`Write complete: ${result.successCount} success, ${result.failCount} failed`);
        
        return result;
    }
    
    chunkArray(array, size) {
        const chunks = [];
        for (let i = 0; i < array.length; i += size) {
            chunks.push(array.slice(i, i + size));
        }
        return chunks;
    }
}

{
    "job": {
        "setting": {
            "speed": {
                "channel": 5
            },
            "errorLimit": {
                "record": 0,
                "percentage": 0.02
            }
        },
        "content": [
            {
                "reader": {
                    "name": "httpreader",
                    "parameter": {
                        "url": "https://api.example.com/data",
                        "method": "GET",
                        "headers": {
                            "Authorization": "Bearer xxx"
                        },
                        "pagination": {
                            "type": "offset",
                            "pageSize": 100,
                            "totalKey": "total",
                            "dataKey": "items"
                        }
                    }
                },
                "writer": {
                    "name": "mysqlwriter",
                    "parameter": {
                        "username": "user",
                        "password": "password",
                        "column": ["id", "name", "email"],
                        "preSql": ["TRUNCATE TABLE target_table"],
                        "connection": [
                            {
                                "jdbcUrl": "jdbc:mysql://localhost:3306/db",
                                "table": ["target_table"]
                            }
                        ]
                    }
                }
            }
        ]
    }
}

{
    "mappings": [
        { "source": "user_name", "target": "name", "transform": "direct" },
        { "source": "user_email", "target": "email", "transform": "direct" },
        { "source": "user_phone", "target": "phone", "transform": "cleanPhone" },
        { "source": "create_time", "target": "createdAt", "transform": "formatDate" },
        { "source": "status_code", "target": "status", "transform": "mapStatus" }
    ],
    "transforms": {
        "cleanPhone": "removeNonNumeric(value)",
        "formatDate": "format(value, 'YYYY-MM-DD HH:mm:ss')",
        "mapStatus": "value === 1 ? 'active' : 'inactive'"
    }
}