ETL方法

@param task 任务名, 对应配置名
@param params etl筛选条件
@return ETL结果
359997
@Override
    public EtlResult etl(String task, List<String> params) {
        EtlResult etlResult = new EtlResult();
        MappingConfig config = rdbMapping.get(task);
        RdbEtlService rdbEtlService = new RdbEtlService(dataSource, config);
        if (config != null) {
                return rdbEtlService.importData(params);
        } else {
            StringBuilder resultMsg = new StringBuilder();
            boolean resSucc = true;
            for (MappingConfig configTmp : rdbMapping.values()) {
                // 取所有的destination为task的配置
                if (configTmp.getDestination().equals(task)) {
                    EtlResult etlRes = rdbEtlService.importData(params);
                    if (!etlRes.getSucceeded()) {
                        resSucc = false;
                        resultMsg.append(etlRes.getErrorMessage()).append("\n");
                    } else {
                        resultMsg.append(etlRes.getResultMessage()).append("\n");
                    }
                }
            }
            if (resultMsg.length() > 0) {
                etlResult.setSucceeded(resSucc);
                if (resSucc) {
                    etlResult.setResultMessage(resultMsg.toString());
                } else {
                    etlResult.setErrorMessage(resultMsg.toString());
                }
                return etlResult;
            }
        }
        etlResult.setSucceeded(false);
        etlResult.setErrorMessage("Task not found");
        return etlResult;
    }
961280
public <T> T query(String sql, RsHandler<T> rsh, Object... params) throws SQLException {
		Connection conn = null;
		try {
			conn = this.getConnection();
			return SqlExecutor.query(conn, sql, rsh, params);
		} catch (SQLException e) {
			throw e;
		} finally {
			this.closeConnection(conn);
		}
	}
442342
@Override
    public Map<String, Object> count(String task) {
        MappingConfig config = rdbMapping.get(task);
        MappingConfig.DbMapping dbMapping = config.getDbMapping();
        String sql = "SELECT COUNT(1) AS cnt FROM " + SyncUtil.getDbTableName(dbMapping);
        Connection conn = null;
        Map<String, Object> res = new LinkedHashMap<>();
        try {
            conn = dataSource.getConnection();
            Util.sqlRS(conn, sql, rs -> {
                try {
                    if (rs.next()) {
                        Long rowCount = rs.getLong("cnt");
                        res.put("count", rowCount);
                    }
                } catch (SQLException e) {
                    logger.error(e.getMessage(), e);
                }
            });
        } catch (SQLException e) {
            logger.error(e.getMessage(), e);
        } finally {
            if (conn != null) {
                try {
                    conn.close();
                } catch (SQLException e) {
                    logger.error(e.getMessage(), e);
                }
            }
        }
        res.put("targetTable", SyncUtil.getDbTableName(dbMapping));

        return res;
    }
514486
@Override
  public Map<String, Object> callExecutorStats(final int executorId, final String action,
      final Pair<String, String>... params) throws IOException, ExecutorManagerException {
    final Executor executor = fetchExecutor(executorId);

    final List<Pair<String, String>> paramList =
        new ArrayList<>();

    // if params = null
    if (params != null) {
      paramList.addAll(Arrays.asList(params));
    }

    paramList
        .add(new Pair<>(ConnectorParams.ACTION_PARAM, action));

    return this.apiGateway.callForJsonObjectMap(executor.getHost(), executor.getPort(),
        "/stats", paramList);
  }
534723
public int execute(String sql, Object... params) throws SQLException {
		Connection conn = null;
		try {
			conn = this.getConnection();
			return SqlExecutor.execute(conn, sql, params);
		} catch (SQLException e) {
			throw e;
		} finally {
			this.closeConnection(conn);
		}
	}
868397
public static int exec(String query, Object ... params){
        return new DB(DB.DEFAULT_NAME).exec(query, params);
    }
1228142
public AsyncTask<Params, Progress, Result> executeTask(Params... params) {
        if (Build.VERSION.SDK_INT >= Build.VERSION_CODES.HONEYCOMB) {
            try {
                return executeOnExecutor(AsyncTask.THREAD_POOL_EXECUTOR, params);
            } catch (Throwable t) {
                return execute(params);
            }
        } else
            return execute(params);
    }
280386
public static SignedCookieData deserialize(String taskParameters) {
        JaxbJsonSerializer<SignedCookieData> serializer =
            new JaxbJsonSerializer<>(SignedCookieData.class);
        try {
            return serializer.deserialize(taskParameters);
        } catch (IOException e) {
            throw new TaskDataException(
                "Unable to create task parameters due to: " + e.getMessage());
        }
    }
316620
public <T> ForkJoinTask<T> submit(ForkJoinTask<T> task) {
        if (task == null)
            throw new NullPointerException();
        doSubmit(task);
        return task;
    }
951769
@Override
  public Object processTask(String taskName, Map<String, String[]> parameterMap) {

    if ("createAdmin".equalsIgnoreCase(taskName)) {
      return userService.createDefaultAdmin();
    }

    return null;

  }
551622
public static ClientResponseImpl transformRestoreParamsToJSON(StoredProcedureInvocation task) {
        Object params[] = task.getParams().toArray();
        if (params.length == 1) {
            try{
                JSONObject jsObj = new JSONObject((String)params[0]);
                String path = jsObj.optString(JSON_PATH);
                String dupPath = jsObj.optString(JSON_DUPLICATES_PATH);
                if(!path.isEmpty() && dupPath.isEmpty()){
                    jsObj.put(JSON_DUPLICATES_PATH, path);
                }
                task.setParams( jsObj.toString() );
            } catch (JSONException e){
                Throwables.propagate(e);
            }
            return null;
        } else if (params.length == 2) {
            if (params[0] == null) {
                return new ClientResponseImpl(ClientResponseImpl.GRACEFUL_FAILURE,
                                              new VoltTable[0],
                                              "@SnapshotRestore parameter 0 was null",
                                              task.getClientHandle());
            }
            if (params[1] == null) {
                return new ClientResponseImpl(ClientResponseImpl.GRACEFUL_FAILURE,
                                              new VoltTable[0],
                                              "@SnapshotRestore parameter 1 was null",
                                              task.getClientHandle());
            }
            if (!(params[0] instanceof String)) {
                return new ClientResponseImpl(ClientResponseImpl.GRACEFUL_FAILURE,
                                              new VoltTable[0],
                                              "@SnapshotRestore param 0 (path) needs to be a string, but was type "
                                              + params[0].getClass().getSimpleName(),
                                              task.getClientHandle());
            }
            if (!(params[1] instanceof String)) {
                return new ClientResponseImpl(ClientResponseImpl.GRACEFUL_FAILURE,
                                              new VoltTable[0],
                                              "@SnapshotRestore param 1 (nonce) needs to be a string, but was type "
                                              + params[1].getClass().getSimpleName(),
                                              task.getClientHandle());
            }
            JSONObject jsObj = new JSONObject();
            try {
                jsObj.put(SnapshotUtil.JSON_PATH, params[0]);
                if (VoltDB.instance().isRunningWithOldVerbs()) {
                    jsObj.put(SnapshotUtil.JSON_PATH_TYPE, SnapshotPathType.SNAP_PATH);
                }
                jsObj.put(SnapshotUtil.JSON_NONCE, params[1]);
                jsObj.put(SnapshotUtil.JSON_DUPLICATES_PATH, params[0]);
            } catch (JSONException e) {
                Throwables.propagate(e);
            }
            task.setParams( jsObj.toString() );
            return null;
        } else {
            return new ClientResponseImpl(ClientResponseImpl.GRACEFUL_FAILURE,
                                          new VoltTable[0],
                                          "@SnapshotRestore supports a single json document parameter or two parameters (path, nonce), " +
                                          params.length + " parameters provided",
                                          task.getClientHandle());
        }
    }
610435
public Task createDocument(Task task) {
        DocumentRevision rev = new DocumentRevision();
        rev.setBody(DocumentBodyFactory.create(task.asMap()));
        try {
            DocumentRevision created = this.mDocumentStore.database().create(rev);
            return Task.fromRevision(created);
        } catch (DocumentException de) {
            return null;
        } catch (DocumentStoreException de) {
            return null;
        }
    }
589788
public CollectionRequest<Task> projects(String task) {
    
        String path = String.format("/tasks/%s/projects", task);
        return new CollectionRequest<Task>(this, Task.class, path, "GET");
    }
1180515
public String performTask(String taskParameters) {
        EnableStreamingTaskParameters taskParams =
            EnableStreamingTaskParameters.deserialize(taskParameters);

        String spaceId = taskParams.getSpaceId();
        boolean secure = taskParams.isSecure();

        log.info("Performing " + TASK_NAME + " task on space " + spaceId +
                 ". Secure streaming set to " + secure);

        // Will throw if bucket does not exist
        String bucketName = unwrappedS3Provider.getBucketName(spaceId);

        String domainName = null;
        String distId = null;
        String oaIdentityId = getOriginAccessId();
        EnableStreamingTaskResult taskResult = new EnableStreamingTaskResult();

        StreamingDistributionSummary existingDist =
            getExistingDistribution(bucketName);

        if (existingDist != null) { // There is an existing distribution
            // Ensure that this is not an attempt to change the security type
            // of this existing distribution
            boolean existingSecure =
                !existingDist.getTrustedSigners().getItems().isEmpty();
            if ((secure && !existingSecure) || (!secure && existingSecure)) {
                throw new UnsupportedTaskException(TASK_NAME,
                                                   "The space " + spaceId + " is already configured to stream as " +
                                                   (secure ? "OPEN" : "SECURE") +
                                                   " and cannot be updated to stream as " +
                                                   (secure ? "SECURE" : "OPEN") +
                                                   ". To do this, you must first execute the " +
                                                   StorageTaskConstants.DELETE_STREAMING_TASK_NAME + " task.");
            }

            distId = existingDist.getId();
            if (!existingDist.isEnabled()) { // Distribution is disabled, enable it
                setDistributionState(distId, true);
            }
            domainName = existingDist.getDomainName();
        } else { // No existing distribution, need to create one
            S3Origin origin = new S3Origin(bucketName + S3_ORIGIN_SUFFIX,
                                           S3_ORIGIN_OAI_PREFIX + oaIdentityId);

            // Only include trusted signers on secure distributions
            TrustedSigners signers = new TrustedSigners();
            if (secure) {
                signers.setItems(Collections.singletonList(cfAccountId));
                signers.setEnabled(true);
                signers.setQuantity(1);
            } else {
                signers.setEnabled(false);
                signers.setQuantity(0);
            }

            StreamingDistribution dist =
                cfClient.createStreamingDistribution(
                    new CreateStreamingDistributionRequest(
                        new StreamingDistributionConfig()
                            .withCallerReference("" + System.currentTimeMillis())
                            .withS3Origin(origin)
                            .withEnabled(true)
                            .withComment("Streaming space: " + spaceId)
                            .withTrustedSigners(signers)))
                        .getStreamingDistribution();
            domainName = dist.getDomainName();
        }

        // Set bucket policy to accept origin access identity
        setBucketAccessPolicy(bucketName, oaIdentityId);

        // Update bucket tags to include streaming host
        Map<String, String> spaceProps =
            s3Provider.getSpaceProperties(spaceId);
        spaceProps.put(STREAMING_HOST_PROP, domainName);
        spaceProps.put(STREAMING_TYPE_PROP,
                       secure ? STREAMING_TYPE.SECURE.name() : STREAMING_TYPE.OPEN.name());
        unwrappedS3Provider.setNewSpaceProperties(spaceId, spaceProps);

        taskResult.setResult("Enable Streaming Task completed successfully");

        // Return results
        taskResult.setStreamingHost(domainName);
        String toReturn = taskResult.serialize();
        log.info("Result of " + TASK_NAME + " task: " + toReturn);
        return toReturn;
    }
1330859
public Long createTask(String taskName, String data, String key1, String key2, Long batchId) {
		FieldValidator.validateTaskData(data);
		FieldValidator.validateTaskKey1(key1);
		FieldValidator.validateTaskKey2(key2);
		TaskConfig tc = context.registry.getTaskConfig(taskName);
		if (tc == null)
			throw new IllegalArgumentException("Task '" + taskName + "' is not known for TED");
		return context.tedDao.createTask(taskName, tc.channel, data, key1, key2, batchId);
	}
1256022
public static JoinPoint<Exception> fromTasks(Task<?,?>... tasks) {
		JoinPoint<Exception> jp = new JoinPoint<>();
		for (Task<?,?> task : tasks) jp.addToJoin(task.getOutput());
		jp.start();
		return jp;
	}
1371557
@Nonnull
    public static <T> List<TtlCallable<T>> gets(@Nullable Collection<? extends Callable<T>> tasks) {
        return gets(tasks, false, false);
    }