DM FLDR JNI 编程指南

8.1 DM FLDR JNI接口介绍

java调用dm快速装载接口导数据是通过jni接口来完成的。所引用的对象在com.dameng.floader.jar中的com.dameng.floader.Instance,com.dameng.floader.Properties。主要类是com.dameng.floader.Instance。

8.2 接口说明

  1. allocInstance

描述:

用于分配快速装载对象实例,所有快速装载都将通过该句柄进行操作。

格式:

boolean

allocInstance();

参数:

返回值:

true或false

注释:

该函数分配出用于快速装载实例句柄,用户必须首先调用该函数获得快速装载的实例句柄,只有通过该句柄才能进行后续的快速装载操作。

2.free

描述:

释放快速装载实例句柄。

格式:

void

free();

参数:

返回值:

注释:

释放快速装载实例句柄,将该实例句柄上所有快速装载所占用的资源释放。

3.setAttribute

描述:

设置快速装载实例的属性。

格式:

bollean

setAttribute(

int attr_id,

String value

);

参数:

attr_id(输入) 属性标识,指出将要设置的属性,关于属性详细内容参见注释部分。

value (输入) 属性值,根据attrid不同该参数的意义不同。

返回值:

true,false

注释:

可设置的属性见下表:

属性(attrid) 属性说明(value)
FLDR_ATTR_SERVER 服务器地址。
FLDR_ATTR_UID 用户名。
FLDR_ATTR_PWD 密码。
FLDR_ATTR_PORT 服务器端口
FLDR_ATTR_TABLE 目标表
FLDR_ATTR_SET_INDENTITY 是否插入自增列(默认FALSE),如果设置为TRUE则服务器将使用指定的数据作为自增列的数据进行插入,如果为FALSE则对于自增列服务器将自动生成数据插入,忽略指定的数据
FLDR_ATTR_DATA_SORTED 数据是否已按照聚集索引排序。默认为FALSE
FLDR_ATTR_INDEX_OPTION 1:不刷新二级索引,数据按照索引先排序,装载完后再将排序的数据插入索引 2:不刷新二级索引,数据装载完成后重建所有二级索引(默认为1)
FLDR_ATTR_DATA_SIZE 指出本地缓冲数据的最大行数。默认为1000,最大为10000,最小为100。
FLDR_ATTR_INSERT_ROWS 8字节整数。只读属性。当前发送到服务器的行数。
FLDR_ATTR_COMMIT_ROWS 8字节整数。只读属性。当前提交到数据库的行数
FLDR_ATTR_PROCESS_ROWS 8字节整数。只读属性。用户绑定的数据行数
FLDR_ATTR_SEND_NODE_NUM 发送链表节点个数,默认128,最大为20480,最小为16。多线程处理中每个线程所支配的向服务器发送数据节点的个数。
FLDR_ATTR_TASK_THREAD_NUM 任务线程数
FLDR_ATTR_BAD_FILE 记录出错数据的文件,出错数据将全部使用字符串表示
FLDR_ATTR_DATA_CHAR_SET 设置输入数据字符类型的字符集
FLDR_ATTR_LOG_FILE 指定日志记录文件
FLDR_ATTR_ERRORS_PERMIT 4字节整数。允许出现错误数据的行数,超过该值装载将报错
FLDR_ATTR_SSL_PATH 在加密通讯模式下加密证书的路径
FLDR_ATTR_SSL_PWD 加密证书的密钥
FLDR_ATTR_LOCAL_CODE 指定本地字符集
FLDR_ATTR_MPP_LOCAL_FLAG 指明是否是mpp模式下的单节点数据。默认值为FALSE
FLDR_ATTR_NULL_MODE 载入NULL字符串时是否作为空值处理,默认值为FALSE
FLDR_ATTR_SERVER_BLDR_NUM 2字节整数。服务器bldr数目,通常多于线程数
FLDR_ATTR_BDTA_SIZE 4字节整数。每次填充数据的行数
FLDR_ATTR_COMPRESS_FLAG 待发送的数据是否先压缩,默认FALSE
FLDR_ATTR_FLUSH_FLAG 指定提交数据时服务器的处理方式。若设置为TRUE则服务器将数据写入磁盘后才返回响应消息;若设置为FALSE则服务器确认数据正确即返回响应消息,之后才将数据写入磁盘

4.getAttribute

描述:

获取快速装载实例的属性。

格式:

String

getAttribute(

int attr_id

);

参数:

attrid(输入) 属性标识,指出将要获取的属性,关于属性详细内容参见setAttribute的注释部分。

返回值:

属性值。

注释:

参见setAttribute函数说明中各属性说明。

5.initializeInstance

描述:

初始化当前快速装载实例。

格式:

boolean

initializeInstance(

String server,

String user,

String pwd,

String table

);

参数:

server (输入) 链接的服务器。将被记录到实例的FLDR_ATTR_SERVER属性中。

user (输入) 登录用户。将被记录到实例的FLDR_ATTR_UID属性中。

pwd(输入) 用户密码。将被记录到实例的FLDR_ATTR_PWD属性中。

table(输出) 进行快速装载的表。将被记录到实例的FLDR_ATTR_TABLE属性中。

返回值:

true,false

注释:

根据用户通过setAttribute设置的属性初始化快速装载环境,如果用户未指定连接则根据用户提供的服务器、用户和密码建立到服务器的连接。

6.unitialize

描述:

释放快速装载实例所占用的资源。

格式:

    boolean
   uninitialize();

参数:

返回值:

true,false

注释:

用于释放所占用快速装载实例所占用的资源。用户在调用该函数后可重新初始化快速装载句柄,进行新的数据装载。

  1. fldr_Bind_Columns_ex

描述:

绑定输入列。

格式:

boolean
fldr_Bind_Columns_ex(
int colNumber, 
int type, 
String datefmt, 
Object data, 
int maxcollen,
int[] ind_len, 
int threadNum
);

参数:

colNumber(输入) 绑定列的编号。从1开始计数,对应快速装载的表的相应列。

type (输入) 用户绑定数据的JAVA类型。

datefmt(输入) 当绑定列对应的表的列为时间日期类型,且绑定的JAVA数据为String类型,该参数为字符串到日期类型的格式串。其他情况,忽略该绑定输入。

data(输入) 指向用于存放数据内存的地址指针。

maxcollen(输入) 用于记录单行数据最大的长度。当进行多行绑定的时候,对于变长的C类型,根据该参数内容进行便宜的计算。对于固定长度的JAVA数据类型,该参数内容被忽略。

ind_len(输入) 用户绑定的数据长度指示符。当为批量绑定时,传入的地址为指向INT类型的数组。如果地址中保存的值为-1,绑定列的该行数据被视为空值。对于BINARY和CHAR变长数据类型,用来指明数据的长度。对于其他定长数据类型,忽略其长度。

threadNum(输入) 多线程调用,线程序号,单线程传0。

返回值:

true,false

注释:

  1. fldr_Send_Rows_ex

描述:

发送行数。

格式:

boolean
fldr_Send_Rows_ex(
long rows, 
int threadNum, 
int seqNo
);

参数:

rows (输入) 在多行绑定的情况,指明绑定的行数。

threadNum(输入) 多线程调用,线程序号,单线程传0。

seqNo(输入) 数据批次,从1开始自增,保证数据有序。

返回值:

true,false

  1. finish

描述:

结束一次批量操作。

格式:

bollean

finish();

参数:

返回值:

true,false

8.3 DM FLDR JNI编程实例

例1 将数据导入到数据库。

package com.dameng;
import java.io.File;
import com.dameng.floader.DataTypes;
import com.dameng.floader.Instance;
import com.dameng.floader.Properties;

public class TestFloder
{

    public static void main(String[] args)
    {
        Instance instance = new Instance();

        int threadNum = 1;

        TestFloder floder = new TestFloder();
        /* ***************************初始化******************************* */
        /* create table sysdba.test1(c1 int ,c2 varchar(50)); */
        String tableName = "SYSDBA.TEST1";
        boolean success = floder.init(instance, threadNum, tableName);
  
        if (success)
        {
            try
            {
            	int		c1[] = {1,2,3,4,5,6,7,8,9};		//数据
            	int		c1_ind[] = {4,4,4,4,4,4,4,4,4};	//长度指示符  -1表示空
            	byte[] 	byteArray = new byte[9];
      
            	for (int i = 0; i < 9; i++)
            	{
            		byteArray[i] = ((Integer)c1[i]).byteValue();
            	}
      
      
            	String	c2[] = {"a1","a12","a123","a1234","a12345","a123456","a1234567","a12345678","a123456789"};
            	int		c2_ind[] = {2,3,4,5,6,7,8,9,10};	//长度指示符  -1表示空
            	byte[][] 	byteArray2 = new byte[9][];
      
            	for (int i = 0; i < 9; i++)
            	{
            		byteArray2[i] = c2[i].getBytes();
            	}     

            	/*
            	fldr_Bind_Columns_ex(
            	int colNumber,			// 列号 从1开始 
            	int type, 				// 列类型从DataTypes取
            	String datefmt, 		// 时间类型格式范式:默认yyyy-mm-dd hh-mm-ss,为空的话,按默认处理
            	Object data, 			// 数值类型直接使数值类型数组,字符串类型是byte数组
            	int maxcollen,			// 数值类型值无效,fldr里面自己按sizeof取长度。,
            							// xxxxxxxxxxxxx注意字符串类型是所有数据总长度xxxxxxxxxxxxxxxxxxx。
            							// fldr里面要按这个总长度申请类型
            	int[] ind_len, 			// 长度指示符,int数组,-1标示空
            							// 数值类型无效,字符串类型标示数据长度。
            	int threadNum			// 多线程下有效,单线程传1
            	)
            	 */
      
            	/*
            	fldr_Send_Rows_ex(
            	long rows,				// 行数 
            	int threadNum, 			// 多线程下有效,单线程传1
            	int seqNo				// 数据批次号,从1开始,必须连续,否则fldr会卡住。
            	)
            	*/
      
            	for (int i = 1; i < 10; i++)
            	{
            		instance.fldr_Bind_Columns_ex(1, DataTypes.FLDR_C_INT, null, c1, 4, c1_ind, 1);
                    instance.fldr_Bind_Columns_ex(2, DataTypes.FLDR_C_CHAR, null, byteArray2, 54, c2_ind, 1);
              
                    instance.fldr_Send_Rows_ex(9, 1, i);
            	}
          

            }
            catch (Exception e)
            {
                e.printStackTrace();
            }
        }
  
  
        /* ***************************释放******************************* */
        if (!success)
        {
            String message = instance.getErrorMsg();
            System.out.println("出错信息:" + message);
            instance.free();
        }
        else
        {
            instance.finish();
            // 返回实际插入的行数
            long rowHaveCopied = Long.parseLong(instance.getAttribute(Properties.FLDR_ATTR_COMMIT_ROWS));
            System.out.println("已复制行数:" + rowHaveCopied);
            instance.unitialize();
            instance.free();
        }

    }

    // 初始化
    public boolean init(Instance instance, int threadNum, String tableName)
    {
        // 分配句柄
        instance.allocInstance();
        // 设置属性
        String host = "localhost";
        String port = "5236";
        String userName = "SYSDBA";
        String password = "SYSDBA";
        instance.setAttribute(Properties.FLDR_ATTR_SERVER, host);
        instance.setAttribute(Properties.FLDR_ATTR_PORT, port);
        instance.setAttribute(Properties.FLDR_ATTR_UID, userName);
        instance.setAttribute(Properties.FLDR_ATTR_PWD, password);

        // 编码
//        instance.setAttribute(Properties.FLDR_ATTR_DATA_CHAR_SET, System.getProperty("file.encoding"));

        // 是否有自增量
//        instance.setAttribute(Properties.FLDR_ATTR_SET_INDENTITY, "0");

        // fldr task线程默认为cpu个数,如果客户端线程数大于cpu个数,则必须设置该参数
//        instance.setAttribute(Properties.FLDR_ATTR_TASK_THREAD_NUM, String.valueOf(threadNum));

        // 一次提交记录数
//        instance.setAttribute(Properties.FLDR_ATTR_DATA_SIZE, "5000");

        // 记录出错的信息
        String dmHome = System.getProperty("DM_HOME");
        if (dmHome != null && !dmHome.equals(""))
        {
            if (dmHome.endsWith("/") || dmHome.endsWith("\\"))
            {
                instance.setAttribute(Properties.FLDR_ATTR_BAD_FILE, dmHome + "BADFILE.TXT");
            }
            else
            {
                instance.setAttribute(Properties.FLDR_ATTR_BAD_FILE, dmHome + File.separator + "BADFILE.TXT");
            }
        }

        // 装载日志
        if (dmHome != null && !dmHome.equals(""))
        {
            if (dmHome.endsWith("/") || dmHome.endsWith("\\"))
            {
                instance.setAttribute(Properties.FLDR_ATTR_LOG_FILE, dmHome + "FLDR_LOG.TXT");
            }
            else
            {
                instance.setAttribute(Properties.FLDR_ATTR_LOG_FILE, dmHome + File.separator + "FLDR_LOG.TXT");
            }
        }

        // 最大容错个数
//        instance.setAttribute(Properties.FLDR_ATTR_ERRORS_PERMIT, String.valueOf(Integer.MAX_VALUE));

        // 是否为mpp
//        instance.setAttribute(Properties.FLDR_ATTR_MPP_CLIENT, "0");

        // 如果是txt到dm8
//            instance.setAttribute(Properties.FLDR_ATTR_COMPRESS_FLAG,  "0");
//            instance.setAttribute(Properties.FLDR_ATTR_DIRECT_MODE,"0");
//            instance.setAttribute(Properties.FLDR_ATTR_DATA_SORTED, "0");
//            instance.setAttribute(Properties.FLDR_ATTR_LOCAL_CODE, "GBK");
//            instance.setAttribute(Properties.FLDR_ATTR_BLOB_TYPE, "0");
//            instance.setAttribute(Properties.FLDR_ATTR_INDEX_OPTION,"1");
//            instance.setAttribute(Properties.FLDR_ATTR_LOB_DIRECTORY, "");
//            instance.setAttribute(Properties.FLDR_ATTR_ERRORS_PERMIT,"1000");
//            instance.setAttribute(Properties.FLDR_ATTR_SKIP_ROWS,"0");
//            instance.setAttribute(Properties.FLDR_ATTR_BUFFER_SIZE,"512");
//            instance.setAttribute(Properties.FLDR_ATTR_SERVER_BLDR_NUM,"512");
//            instance.setAttribute(Properties.FLDR_ATTR_READ_ROWS,"512");
//            instance.setAttribute(Properties.FLDR_ATTR_ROWS_COMMIT,"512");
//            instance.setAttribute(Properties.FLDR_ATTR_BDTA_SIZE,"512");

        boolean success = instance.initializeInstance(null, null, null, tableName);
        return success;
    }

}

例2 将数据库表中的数据导出到文件中

第一步,数据准备,创建数据库表test。

create table test(c1 int);

insert into test values(1),(2),(3);

commit;

第二步,将表test中的数据导出到文件e:\fldr_out.txt中。

package com.dameng;
import com.dameng.floader.Instance;
import com.dameng.floader.Properties;

public class TestFloder {
	public static void main(String[] args)
    {
		Instance instance = new Instance();
		try
        {
             //初始化
            if (instance.allocInstance())
            {
                instance.setAttribute(Properties.FLDR_ATTR_SERVER, "192.168.1.55");
                instance.setAttribute(Properties.FLDR_ATTR_PORT, "5236");
                instance.setAttribute(Properties.FLDR_ATTR_UID, "SYSDBA");
                instance.setAttribute(Properties.FLDR_ATTR_PWD, "SYSDBA");
                instance.setAttribute(Properties.FLDR_ATTR_DATA_CHAR_SET, "GBK");
          
                instance.initializeInstance(null, null, null, "\"SYSDBA\".\"TEST\"");
                String ctl = "OPTIONS\r\n" + 
                        "(\r\n" + 
                        "mode='out'\r\n" + 
                        ")\r\n" + 
                        "Load\r\n" + 
                        "infile 'e:\\fldr_out.txt'\r\n" + 
                        "Append\r\n" + 
                        "into table \"SYSDBA\".\"TEST\"\r\n" + 
                        "FIELDS '|'";
                //导出的ctrl内容
                instance.setControl(ctl);
            }
            else
            {
                System.out.println(instance.getErrorMsg());
            }
        }
        finally
        {
            //释放
            instance.unitialize();
            instance.free();
        }
    }
}

第三步,查看文件e:\fldr_out.txt。

1

2

3
微信扫码
分享文档
扫一扫
联系客服