输出到OSS的非结构化数据

通过外部表关联OSS存储路径后，可以对外部表进行INSERT OVERWRITE/INSERT INTO操作，即可将数据输出到OSS。SQL语法如下。

INSERT OVERWRITE|INTO TABLE <external_tablename> [PARTITION (partcol1=val1, partcol2=val2 ...)]
select_statement
FROM <from_tablename> 
[WHERE where_condition];                         

输出逻辑都必须实现Outputer接口 。

package com.aliyun.odps.examples.unstructured.text;
import com.aliyun.odps.data.Record;
import com.aliyun.odps.io.OutputStreamSet;
import com.aliyun.odps.io.SinkOutputStream;
import com.aliyun.odps.udf.DataAttributes;
import com.aliyun.odps.udf.ExecutionContext;
import com.aliyun.odps.udf.Outputer;
import java.io.IOException;
public class TextOutputer extends Outputer {
    private SinkOutputStream outputStream;
    private DataAttributes attributes;
    private String delimiter;
    public TextOutputer (){
        // default delimiter, this can be overwritten if a delimiter is provided through the attributes.
        this.delimiter = "|";
    }
    @Override
    public void output(Record record) throws IOException {
        this.outputStream.write(recordToString(record).getBytes());
    }
    // no particular usage of execution context in this example
    @Override
    public void setup(ExecutionContext ctx, OutputStreamSet outputStreamSet, DataAttributes attributes) throws IOException {
        this.outputStream = outputStreamSet.next();
        this.attributes = attributes;
        this.delimiter = this.attributes.getValueByKey("delimiter");
        if ( this.delimiter == null)
        {
            this.delimiter=",";
        }
        System.out.println("Extractor using delimiter [" + this.delimiter + "].");
    }
    @Override
    public void close() {
        // no-op
    }
    private String recordToString(Record record){
        StringBuilder sb = new StringBuilder();
        for (int i = 0; i < record.getColumnCount(); i++)
        {
            if (null == record.get(i)){
                sb.append("NULL");
            }
            else{
                sb.append(record.get(i).toString());
            }
            if (i != record.getColumnCount() - 1){
                sb.append(this.delimiter);
            }
        }
        sb.append("\n");
        return sb.toString();
    }
}

通常，数据处理发生在output(Record)这个接口内。MaxCompute系统根据当前Outputer分配处理的每个输入Record调用一次output(Record)。假设，在一个output(Record)调用返回的时候，代码已经消费完这个Record，则在当前output(Record)返回后，系统会将Record所使用的内存作它用。因此，当Record中的信息在跨多个output()函数调用时，需要调用当前处理的record.clone()方法，将当前Record保存下来。

Exatractor用于数据的读入、解析、处理等，如果输出的表最终不需要再通过MaxCompute进行读取等，可以无需定义。

package com.aliyun.odps.examples.unstructured.text;
import com.aliyun.odps.Column;
import com.aliyun.odps.data.ArrayRecord;
import com.aliyun.odps.data.Record;
import com.aliyun.odps.io.InputStreamSet;
import com.aliyun.odps.udf.DataAttributes;
import com.aliyun.odps.udf.ExecutionContext;
import com.aliyun.odps.udf.Extractor;
import java.io.BufferedReader;
import java.io.IOException;
import java.io.InputStream;
import java.io.InputStreamReader;
/**
 * Text extractor that extract schematized records from formatted plain-text(csv, tsv etc.)
 **/
public class TextExtractor extends Extractor {
    private InputStreamSet inputs;
    private String columnDelimiter;
    private DataAttributes attributes;
    private BufferedReader currentReader;
    private boolean firstRead = true;
    public TextExtractor() {
        // default to ",", this can be overwritten if a specific delimiter is provided (via DataAttributes)
        this.columnDelimiter = ",";
    }
    // no particular usage for execution context in this example
    @Override
    public void setup(ExecutionContext ctx, InputStreamSet inputs, DataAttributes attributes) {
        this.inputs = inputs;
        this.attributes = attributes;
        // check if "delimiter" attribute is supplied via SQL query
        String columnDelimiter = this.attributes.getValueByKey("delimiter");
        if ( columnDelimiter != null)
        {
            this.columnDelimiter = columnDelimiter;
        }
        System.out.println("TextExtractor using delimiter [" + this.columnDelimiter + "].");
        // note: more properties can be inited from attributes if needed
    }
    @Override
    public Record extract() throws IOException {
        String line = readNextLine();
        if (line == null) {
            return null;
        }
        return textLineToRecord(line);
    }
    @Override
    public void close(){
        // no-op
    }
    private Record textLineToRecord(String line) throws IllegalArgumentException
    {
        Column[] outputColumns = this.attributes.getRecordColumns();
        ArrayRecord record = new ArrayRecord(outputColumns);
        if (this.attributes.getRecordColumns().length != 0){
            // string copies are needed, not the most efficient one, but suffice as an example here
            String[] parts = line.split(columnDelimiter);
            int[] outputIndexes = this.attributes.getNeededIndexes();
            if (outputIndexes == null){
                throw new IllegalArgumentException("No outputIndexes supplied.");
            }
            if (outputIndexes.length != outputColumns.length){
                throw new IllegalArgumentException("Mismatched output schema: Expecting "
                        + outputColumns.length + " columns but get " + parts.length);
            }
            int index = 0;
            for(int i = 0; i < parts.length; i++){
                // only parse data in columns indexed by output indexes
                if (index < outputIndexes.length && i == outputIndexes[index]){
                    switch (outputColumns[index].getType()) {
                        case STRING:
                            record.setString(index, parts[i]);
                            break;
                        case BIGINT:
                            record.setBigint(index, Long.parseLong(parts[i]));
                            break;
                        case BOOLEAN:
                            record.setBoolean(index, Boolean.parseBoolean(parts[i]));
                            break;
                        case DOUBLE:
                            record.setDouble(index, Double.parseDouble(parts[i]));
                            break;
                        case DATETIME:
                        case DECIMAL:
                        case ARRAY:
                        case MAP:
                        default:
                            throw new IllegalArgumentException("Type " + outputColumns[index].getType() + " not supported for now.");
                    }
                    index++;
                }
            }
        }
        return record;
    }
    /**
     * Read next line from underlying input streams.
     * @return The next line as String object. If all of the contents of input
     * streams has been read, return null.
     */
    private String readNextLine() throws IOException {
        if (firstRead) {
            firstRead = false;
            // the first read, initialize things
            currentReader = moveToNextStream();
            if (currentReader == null) {
                // empty input stream set
                return null;
            }
        }
        while (currentReader != null) {
            String line = currentReader.readLine();
            if (line != null) {
                return line;
            }
            currentReader = moveToNextStream();
        }
        return null;
    }
    private BufferedReader moveToNextStream() throws IOException {
        InputStream stream = inputs.next();
        if (stream == null) {
            return null;
        } else {
            return new BufferedReader(new InputStreamReader(stream));
        }
    }
}

将自定义代码编译打包，并作为Jar资源上传到MaxCompute。例如Jar包名为odps-TextStorageHandler.jar，则上传为MaxCompute Resource的示例如下。

add jar odps-TextStorageHandler.jar;

与使用内置StorageHandler类似，需要建立一个外部表，不同的是这次需要指定数据输出到外部表时候，使用自定义的StorageHandler。

CREATE EXTERNAL TABLE IF NOT EXISTS output_data_txt_external
(
vehicleId int,
recordId int,
patientId int,
calls int,
locationLatitute double,
locationLongtitue double,
recordTime string,
direction string
)
STORED BY 'com.aliyun.odps.examples.unstructured.text.TextStorageHandler' 
WITH SERDEPROPERTIES(
    'delimiter'='|'
    [,'odps.properties.rolearn'='${roleran}'])
LOCATION 'oss://${endpoint}/${bucket}/${userfilePath}/'
USING 'odps-TextStorageHandler.jar';

通过自定义StorageHandler创建External Table关联上OSS存储路径后，可以对External Table进行标准的SQL的INSERT OVERWRITE/INTO操作，将数据输出到OSS的方式与内置StorageHandler一样。

INSERT OVERWRITE|INTO TABLE <external_tablename> [PARTITION (partcol1=val1, partcol2=val2 ...)]
select_statement
FROM <from_tablename> 
[WHERE where_condition];