Create Import and Export Tasks

Create connectors in a workspace and use them to submit file import or export tasks. The creation invocation returns a task resource; use this resource to read task details, execution runs, or statuses before deciding whether to adjust, retry, or delete the task.

Task workflow

  1. The caller selects the connector configuration and transmission task inputs in the target workspace.

  2. The SDK creates or binds the connector, then uses it to submit an import or export task.

  3. The SDK returns the corresponding task resource.

  4. Use this resource to read task details, execution runs, task files, or running statuses.

Prerequisites

Required item

Role on this page

Authenticated client and selected workspace

Define the scope of connectors and tasks.

Connector name, source type, and usage type

Create the connector.

Import or export configuration

Define data sources, target locations, and processing rules.

Target volume or files to export

Specify transmission destinations or payload content.

External system access criteria

Enable the connector to access the selected external system.

Import and export configurations are provided by the caller. The SDK does not infer source, target, or overwrite policies from file paths, display names, or local environments.

Create a file import task

The following function creates a connector and submits a file import task. Successful submission only indicates that the task creation request has returned, not that the data has finished importing.

import moi_product_sdk as sdk


def create_file_import_task(
    workspace,
    connector_name,
    source_type,
    usage_type,
    config_type,
    task_name,
    volume_id,
    source_uris,
    load_mode_config,
    file_filter_config,
):
    connector, _ = workspace.create_connector(
        connector_name, source_type, usage_type
    )
    task, _ = connector.create_file_import_task(
        sdk.FileImportTaskSpec(
            config_type,
            source_uris,
            load_mode_config,
            file_filter_config,
            name=task_name,
            volume_id=volume_id,
        )
    )
    return connector, task

import (
	"context"

	sdk "github.com/matrixorigin/matrixflow/sdk/go-sdk"
)

func createFileImportTask(
	ctx context.Context,
	workspace *sdk.WorkspaceHandle,
	connectorName string,
	sourceType, usageType, configType int,
	taskName, volumeID string,
	sourceURIs []string,
	loadModeConfig, fileFilterConfig map[string]any,
) (*sdk.ConnectorHandle, *sdk.ImportTaskHandle, error) {
	connector, _, err := workspace.CreateConnector(ctx, connectorName, sourceType, usageType)
	if err != nil {
		return nil, nil, err
	}
	task, _, err := connector.CreateFileImportTask(ctx, sdk.FileImportTaskSpec{
		ConfigType:       configType,
		Name:             taskName,
		VolumeID:         volumeID,
		URIs:             sourceURIs,
		LoadModeConfig:   loadModeConfig,
		FileFilterConfig: fileFilterConfig,
	})
	if err != nil {
		return nil, nil, err
	}
	return connector, task, nil
}

Retain the import task resource returned by the function. If network timeouts occur, query known tasks first rather than directly resubmitting duplicate requests.

Inspect an import task

Use the import task resource returned from the previous step to read details, execution runs, and task files. They reflect the task information, execution history, and file list retrieved at this request; do not mistake task creation response for transfer completion.

details = import_task.info()
runs = import_task.runs()
files = import_task.files()

details, err := importTask.Info(ctx)
if err != nil {
	return err
}
runs, err := importTask.Runs(ctx)
if err != nil {
	return err
}
files, err := importTask.Files(ctx)
if err != nil {
	return err
}
_ = details
_ = runs
_ = files

Create an export task

Export tasks require a task name, export type, at least one target configuration, and at least one file to export. File resource IDs and full paths must originate from known resources or caller selection. You can create an export task using the connector resource returned from the previous step.

export_task, _ = connector.create_export_task_from_spec(
    sdk.ExportTaskSpec(
        "export-orders",
        "oss",
        2,
        sdk.ExportTaskConfig(s3_config={"path": "/exports/orders"}),
        [sdk.ExportTaskFile("file-1", ["sales", "orders"], is_raw=True)],
    )
)

isRaw := true
exportTask, _, err := connector.CreateExportTaskFromSpec(ctx, sdk.ExportTaskSpec{
	Name:          "export-orders",
	ConnectorName: "oss",
	Type:          2,
	Config:         sdk.ExportTaskConfig{S3Config: map[string]any{"path": "/exports/orders"}},
	Files: []sdk.ExportTaskFile{{
		FileID:   "file-1",
		FullPath: []string{"sales", "orders"},
		IsRaw:    &isRaw,
	}},
})
if err != nil {
	return err
}
_ = exportTask

Result confirmation

Import tasks provide details, execution runs, and task files; export tasks provide details, task files, and run statuses. Before retrying, pausing, resuming, updating, or deleting, check current task information and data impact.

Limitations

  • Task creation indicates request submission, not completed data import or export.

  • Exports may write data to external storage. Before creating, retrying, or deleting, verify target locations, access criteria, and overwrite policies.

  • Before retrying, check the failure cause and whether data has been partially written to avoid duplicate processing.

Next steps

Last updated on