Examples

These examples use the ObjectStorageSource operator.

a) ObjectStorageSource with dynamic object names to be read

Sample is using bucket as submission parameter and cos application configuration with property cos.creds to specify the IBM COS credentials: As endpoint is the public us-geo (CROSS REGION) the default value of the os-endpoint submission parameter.


composite Main {
    param
        expression<rstring> $bucket: getSubmissionTimeValue("os-bucket");
        expression<rstring> $endpoint: getSubmissionTimeValue("os-endpoint", "s3.us.cloud-object-storage.appdomain.cloud");
    graph
        // ObjectStorageScan operator with directory and pattern
        stream<rstring objectname> Scanned = com.teracloud.streams.objectstorage::ObjectStorageScan() {
            param

                objectStorageURI: com.teracloud.streams.objectstorage.s3::getObjectStorageURI($bucket);
                endpoint: $endpoint;
                directory: "/sample";
                pattern: ".*";
        }

        // use a ObjectStorageSource operator to process the object names
        // The first output attribute must be the data attribute.
        // Attributes from input stream, like objectname, are forwarded if defined in output stream with same attribute name and type.
        stream<rstring line, rstring objectname> Data = com.teracloud.streams.objectstorage::ObjectStorageSource(Scanned) {
            param
                objectStorageURI: com.teracloud.streams.objectstorage.s3::getObjectStorageURI($bucket);
                endpoint: $endpoint;
        }
}

b) ObjectStorageSource with static object name to be read

Sample is using the credentials parameter to specify the IBM COS credentials. Set the objectStorageURI either in format "cos://<bucket-name>/" or "s3a://<bucket-name>/".


composite Main {
    param
        expression<rstring> $credentials: getSubmissionTimeValue("os-credentials");
        expression<rstring> $objectStorageURI: getSubmissionTimeValue("os-uri");
        expression<rstring> $endpoint: getSubmissionTimeValue("os-endpoint", "s3.us.cloud-object-storage.appdomain.cloud");
    graph
        // read text object
        // use a ObjectStorageSource operator with no input port to process a single object
        stream<rstring line> TxtData = com.teracloud.streams.objectstorage::ObjectStorageSource() {
            param
                credentials: $credentials;
                objectStorageURI: $objectStorageURI;
                endpoint: $endpoint;
                objectName: "sample.txt";
        }

        // read binary object
        // use a ObjectStorageSource operator with no input port to process a single object
        stream<blob block> BinData = com.teracloud.streams.objectstorage::ObjectStorageSource() {
            param
                credentials: $credentials;
                objectStorageURI: $objectStorageURI;
                endpoint: $endpoint;
                objectName: "sample.bin";
                blockSize: 0; // loads file as a single tuple
        }
}