kettle插件开发，json转string 格式 -

caiyeyu

浏览: 21594 次
性别:
来自: 杭州

最近访客更多访客>>

Crazysuperlittleboy

tianxiao888

heysummer

风中的一匹狼tb

博主相关

博客

微博

相册

留言

关于我

文章分类

社区版块

存档分类

2016-02 ( 9)
更多存档...

kettle插件开发，json转string 格式

博客分类：

kettle插件开发

kettle插件开发 json转string 插件开发 kettle etl

基于kettle版本5.4.0.1-130
1：实现类

2:JSONToStringMeta
说明：配置标签@Step(),初始化程序
extends BaseStepMeta :继承BaseStepMeta
implements StepMetaInterface :实现接口StepMetaInterface
整个代码的执行过程是：先运行配置标签meta类型,通过meta类找到Dialog类，当运行脚本时调用实现类

@Step(id="bsoft-json", image="JSO.png", name="JSONToString", description="json格式转string", categoryDescription="Bsoft-Dc")

public class JSONToStringMeta extends BaseStepMeta implements StepMetaInterface {
	private XMLField[] inputFields;
	private ValueMetaAndData value;
	private String outputName;

	public JSONToStringMeta() {

		super(); // allocate BaseStepInfo
	}

	//流程中新增列字段
	@SuppressWarnings("deprecation")
	@Override
	public void getFields(RowMetaInterface r, String origin, RowMetaInterface[] info, StepMeta nextStep, VariableSpace space) { 
		ValueMetaInterface v = new ValueMeta(outputName);
		v.setType(ValueMetaInterface.TYPE_STRING);
		r.addValueMeta(v);
	}

	@Override
	public Object clone() {
		JSONToStringMeta retval = (JSONToStringMeta) super.clone();
		int nrfields = inputFields.length;
		retval.allocateInput(nrfields);
		for (int i = 0; i < nrfields; i++) {
			retval.inputFields[i] = (XMLField) inputFields[i].clone();
		}

		return retval;
	}

	public void allocateInput(int nrfields) {
		inputFields = new XMLField[nrfields];
	}

	//读取ktr脚本的xml数据
	@Override
	public void loadXML(Node stepnode, List<DatabaseMeta> databases, Map<String, Counter> counters) throws KettleXMLException {
		outputName = XMLHandler.getTagValue(stepnode, "outputName"); //$NON-NLS-1$ 

		Node infields = XMLHandler.getSubNode(stepnode, "inFields"); //$NON-NLS-1$

		int num = XMLHandler.countNodes(infields, "field"); //$NON-NLS-1$

		allocateInput(num);

		for (int i = 0; i < num; i++) {
			Node fnode = XMLHandler.getSubNodeByNr(infields, "field", i); //$NON-NLS-1$ 
			inputFields[i] = new XMLField();
			inputFields[i].setType(XMLHandler.getTagValue(fnode, "type")); //$NON-NLS-1$  
			inputFields[i].setFieldName(XMLHandler.getTagValue(fnode, "name")); //$NON-NLS-1$  
			inputFields[i].setFormat(XMLHandler.getTagValue(fnode, "format")); //$NON-NLS-1$ 
			inputFields[i].setLength(Const.toInt( XMLHandler.getTagValue(fnode, "length"), -1)); //$NON-NLS-1$
			inputFields[i].setPrecision(Const.toInt( XMLHandler.getTagValue(fnode, "precision"), -1)); //$NON-NLS-1$ 

		}

	}

	//将对象Meta数据转成xml返回
	@Override
	public String getXML() {
		StringBuffer retval = new StringBuffer(300);
		retval.append("    " + XMLHandler.addTagValue("outputName", outputName == null ? "" : outputName));

		XMLField[] infields = getInputField();

		retval.append("    <inFields>"); //$NON-NLS-1$
		for (int i = 0; i < infields.length; i++) {
			inputFields[i] = infields[i];
			retval.append("      <field>"); //$NON-NLS-1$ 
			retval.append("        ").append(XMLHandler.addTagValue("name", inputFields[i].getFieldName())); //$NON-NLS-1$ //$NON-NLS-2$ 
			retval.append("        ").append(XMLHandler.addTagValue("type", inputFields[i].getTypeDesc())); //$NON-NLS-1$ //$NON-NLS-2$
			retval.append("        ").append(XMLHandler.addTagValue("format", inputFields[i].getFormat())); //$NON-NLS-1$ //$NON-NLS-2$ 
			retval.append("        ").append(XMLHandler.addTagValue("length", inputFields[i].getLength())); //$NON-NLS-1$ //$NON-NLS-2$
			retval.append("        ").append(XMLHandler.addTagValue("precision", inputFields[i].getPrecision())); //$NON-NLS-1$ //$NON-NLS-2$ 
			retval.append("      </field>"); //$NON-NLS-1$
		}
		retval.append("    </inFields>"); //$NON-NLS-1$ 

		return retval.toString();
	}
//界面调试
	@Override
	public void check(List<CheckResultInterface> remarks, TransMeta transmeta, StepMeta stepMeta, RowMetaInterface prev, String input[], String output[], RowMetaInterface info) {

		CheckResult cr;

		if (prev == null || prev.size() == 0) {

			cr = new CheckResult(CheckResultInterface.TYPE_RESULT_WARNING,

			"Not receiving any fields from previous steps!", stepMeta);

			remarks.add(cr);

		}

	}
//获得缓存中的数据
	@Override
	public void readRep(Repository rep, ObjectId id_step, List<DatabaseMeta> databases, Map<String, Counter> counters) throws KettleStepException {

		try {
			outputName = rep.getStepAttributeString(id_step, "outputName");
			int inFields = rep.countNrStepAttributes(id_step, "inFields"); //$NON-NLS-1$
			allocateInput(inFields);
			for (int i = 0; i < inFields; i++) {
				inputFields[i] = new XMLField();
				inputFields[i].setFieldName(rep.getStepAttributeString(id_step,
						i, "field_name")); //$NON-NLS-1$ 
				inputFields[i].setType(rep.getStepAttributeString(id_step, i,
						"field_type")); //$NON-NLS-1$
				inputFields[i].setFormat(rep.getStepAttributeString(id_step, i,
						"field_format")); //$NON-NLS-1$ 
				inputFields[i].setLength((int) rep.getStepAttributeInteger(
						id_step, i, "field_length")); //$NON-NLS-1$
				inputFields[i].setPrecision((int) rep.getStepAttributeInteger(
						id_step, i, "field_precision")); //$NON-NLS-1$ 
			}

		} catch (KettleException e) {
			throw new KettleStepException(e.getMessage());
		}

	}

	//保存数据到缓存中
	@Override
	public void saveRep(Repository rep, ObjectId id_transformation, ObjectId id_step) throws KettleException {
		try {
			rep.saveStepAttribute(id_transformation, id_step, "outputName", outputName);

			XMLField[] nrfields = getInputFields();

			for (int i = 0; i < nrfields.length; i++) {
				rep.saveStepAttribute(id_transformation, id_step, i,
						"field_name", nrfields[i].getFieldName()); //$NON-NLS-1$ 
				rep.saveStepAttribute(id_transformation, id_step, i,
						"field_type", nrfields[i].getTypeDesc()); //$NON-NLS-1$
				rep.saveStepAttribute(id_transformation, id_step, i,
						"field_format", nrfields[i].getFormat()); //$NON-NLS-1$ 
				rep.saveStepAttribute(id_transformation, id_step, i,
						"field_length", nrfields[i].getLength()); //$NON-NLS-1$
				rep.saveStepAttribute(id_transformation, id_step, i,
						"field_precision", nrfields[i].getPrecision()); //$NON-NLS-1$ 

			}

		} catch (KettleException e) {
			throw new KettleStepException(e.getMessage());
		}
	}

	/**
	 * @return inputFields
	 */
	public XMLField[] getInputFields() {
		return inputFields;
	}

	/**
	 * @param inputFields
	 *            inputFields
	 */
	public void setInputFields(XMLField[] inputFields) {
		this.inputFields = inputFields;
	}

	@Override
	public void setDefault() {
		int nrfields = 0;
		allocateInput(nrfields);
		XMLField[] xml = new XMLField[nrfields];
		for (int i = 0; i < nrfields; i++) {
			XMLField inPutField = new XMLField();
			xml[i] = inPutField;
		}
		setInputField(xml);

	}

	public StepDialogInterface getDialog(Shell shell, StepMetaInterface meta,
			TransMeta transMeta, String name) {

		return new JSONToStringDialog(shell, meta, transMeta, name);

	}

	@Override
	public StepInterface getStep(StepMeta stepMeta,
			StepDataInterface stepDataInterface, int cnr, TransMeta transMeta,
			Trans disp) {

		return new JSONToString(stepMeta, stepDataInterface, cnr, transMeta,
				disp);

	}

	@Override
	public StepDataInterface getStepData() {

		return new JSONToStringData();

	}

	//控制错误输出
	@Override
	public boolean supportsErrorHandling() {
		return true;
	}

	/**
	 * @return url
	 */
	public String getOutputName() {
		return outputName;
	}

	/**
	 * @param url
	 *            url
	 */
	public void setOutputName(String packageName) {
		this.outputName = packageName;
	}

	/**
	 * @return filterField
	 */
	public XMLField[] getInputField() {
		return inputFields;
	}

	/**
	 * @param filterField
	 *            filterField
	 */
	public void setInputField(XMLField[] inputFields) {
		this.inputFields = inputFields;
	}

	public ValueMetaAndData getValue() {

		return value;
	}

	public void setValue(ValueMetaAndData value) {

		this.value = value;
	}

}

3：JSONToStringData：定义共享对象类

public class JSONToStringData extends BaseStepData implements
		StepDataInterface {
	public RowMetaInterface outputRowMeta;  
	public SAXReader saxReader=null;
	public Configuration conf =null;
	public List<String> typeList=null;
	public FileSystem fs=null;
	public JSONToStringData() {
		super();  
		saxReader = new SAXReader();
		conf= new Configuration();
		typeList=new ArrayList<String>();
	}
 
}

4：JSONToStringDialog：kettle的ui管理界面
说明：extends BaseStepDialog implements StepDialogInterface

public class JSONToStringDialog extends BaseStepDialog implements StepDialogInterface {
	private static Class<?> PKG = JSONToStringMeta.class; // for i18n
	private JSONToStringMeta input;
 
	 
	private ModifyListener lsMod;
	
	private CTabFolder   wTabFolder;
	private FormData     fdTabFolder;
	
	private CTabItem      inputTab,outputTab,dbTab;
	private MessageBox msgError; 
	 

	private Label className_lable;
	private CCombo  className_text;  
	private FormData className_lform, className_tform;
	
	private Label input_lable;
	private TableView input_view;
	private FormData input_lform, input_tform; 
	 
	
	public JSONToStringDialog(Shell parent, Object in, TransMeta transMeta, String sname)
	{
		super(parent, (BaseStepMeta) in, transMeta, sname); 
		input = (JSONToStringMeta) in; 
	}

	@Override
	public String open()

	{// 初始化控制台

		Shell parent = getParent();
		Display display = parent.getDisplay();
		
		shell = new Shell(parent, SWT.DIALOG_TRIM | SWT.RESIZE | SWT.MAX | SWT.MIN);
 		props.setLook(shell);
        setShellImage(shell, input);
 		
		lsMod = new ModifyListener() 
		{
			@Override
			public void modifyText(ModifyEvent e) 
			{
				input.setChanged();
			}
		};
		changed = input.hasChanged();

		FormLayout formLayout = new FormLayout ();
		formLayout.marginWidth  = Const.FORM_MARGIN;
		formLayout.marginHeight = Const.FORM_MARGIN;

		shell.setLayout(formLayout);
		shell.setText("json转string"); //$NON-NLS-1$
		
		int middle = props.getMiddlePct();
		int margin = Const.MARGIN;
		
 

		// 参数输入
		wlStepname=new Label(shell, SWT.RIGHT);
		wlStepname.setText("名称"); //$NON-NLS-1$
		props.setLook(wlStepname);
		fdlStepname=new FormData();
		fdlStepname.left = new FormAttachment(0, 0);
		fdlStepname.right = new FormAttachment(middle, -margin);
		fdlStepname.top = new FormAttachment(0, 0);
		wlStepname.setLayoutData(fdlStepname);
		wStepname=new Text(shell, SWT.SINGLE | SWT.LEFT | SWT.BORDER);
		wStepname.setText(stepname);
 		props.setLook(wStepname);
		wStepname.addModifyListener(lsMod);
		fdStepname=new FormData();
		fdStepname.left = new FormAttachment(middle, 0);
		fdStepname.top  = new FormAttachment(0, margin);
		fdStepname.right= new FormAttachment(100, 0);
		wStepname.setLayoutData(fdStepname);
		
		wTabFolder = new CTabFolder(shell, SWT.BORDER);
 		props.setLook(wTabFolder, Props.WIDGET_STYLE_TAB);
 		wTabFolder.setSimple(false);
 		
		FormLayout fieldsLayout = new FormLayout ();
		fieldsLayout.marginWidth  = Const.FORM_MARGIN;
		fieldsLayout.marginHeight = Const.FORM_MARGIN;
		
		Composite inputComp = new Composite(wTabFolder, SWT.NONE);
		inputComp.setLayout(fieldsLayout);
 		props.setLook(inputComp);
 		
 		inputTab = new CTabItem(wTabFolder, SWT.NONE);
 		inputTab.setText("参数输入");
		
 		inputComp.layout();
		inputTab.setControl(inputComp);

		fdTabFolder = new FormData();
		fdTabFolder.left  = new FormAttachment(0, 0);
		fdTabFolder.top   = new FormAttachment(wStepname, margin);
		fdTabFolder.right = new FormAttachment(100, 0);
		fdTabFolder.bottom= new FormAttachment(100, -50);
		wTabFolder.setLayoutData(fdTabFolder); 
		   
		//输入包路径
		className_lable = new Label(inputComp, SWT.RIGHT);
		className_lable.setText("输出字段:"); 
		props.setLook(className_lable);
		className_lform = new FormData();
		className_lform.left = new FormAttachment(0, 0);
		className_lform.right = new FormAttachment(middle, -margin);
		className_lform.top = new FormAttachment(wOK, margin);
		className_lable.setLayoutData(className_lform); 
		className_text = new CCombo(inputComp, SWT.SINGLE | SWT.LEFT | SWT.BORDER);
		props.setLook(className_text);
		className_text.addModifyListener(lsMod);
		className_tform = new FormData();
		className_tform.left = new FormAttachment(middle, 0);
		className_tform.right = new FormAttachment(100, 0);
		className_tform.top = new FormAttachment(wOK, margin);
		className_text.setLayoutData(className_tform);  
	    
		input_lable=new Label(inputComp, SWT.NONE);
		input_lable.setText("输入参数配置:"); //$NON-NLS-1$
		props.setLook(input_lable);
		input_lform=new FormData();
		input_lform.left  = new FormAttachment(0, 0);
		input_lform.right = new FormAttachment(100, 0);
		input_lform.top   = new FormAttachment(className_text, -margin+2);
		input_lable.setLayoutData(input_lform);
		
		int FieldsRows=0;
		if(input.getInputField()!=null){
			FieldsRows=input.getInputField().length;
		}
		
		//
		ColumnInfo[] colinf = new ColumnInfo[] {
				new ColumnInfo("字段", ColumnInfo.COLUMN_TYPE_CCOMBO, setRowMeta()), //$NON-NLS-1$
				new ColumnInfo("改名为", ColumnInfo.COLUMN_TYPE_TEXT, false), //$NON-NLS-1$
				new ColumnInfo("类型", ColumnInfo.COLUMN_TYPE_CCOMBO, ValueMeta.getTypes()), //$NON-NLS-1$
				new ColumnInfo("长度", ColumnInfo.COLUMN_TYPE_TEXT, false), //$NON-NLS-1$
				new ColumnInfo("精度", ColumnInfo.COLUMN_TYPE_TEXT, false), //$NON-NLS-1$ 
			}; 
		
		input_view = new TableView(transMeta, inputComp, SWT.BORDER
				| SWT.FULL_SELECTION | SWT.MULTI | SWT.V_SCROLL | SWT.H_SCROLL, colinf, FieldsRows, lsMod,
				props);

		input_tform = new FormData();
		input_tform.left = new FormAttachment(0, 0);
		input_tform.top = new FormAttachment(input_lable, margin);
		input_tform.right = new FormAttachment(100, 0);
		input_tform.bottom = new FormAttachment(100, 0);
		input_view.setLayoutData(input_tform);
		 
 
		wOK=new Button(shell, SWT.PUSH);
		wOK.setText(BaseMessages.getString(PKG, "System.Button.OK"));

		wCancel=new Button(shell, SWT.PUSH);
		wCancel.setText(BaseMessages.getString(PKG, "System.Button.Cancel"));

		setButtonPositions(new Button[] { wOK, wCancel }, margin, wTabFolder);
		
		msgError = new MessageBox(shell, SWT.ICON_ERROR);
		
		setSize();
		initInfo();
		Listener lsCancel = new Listener() {
			@Override
			public void handleEvent(Event e) {
				cancel();
			}
		};

		Listener lsOk = new Listener() {
			@Override
			public void handleEvent(Event e) {
				ok();
			}
		}; 
		
		wCancel.addListener(SWT.Selection, lsCancel);
		
		wOK.addListener(SWT.Selection, lsOk);  
		
		wTabFolder.setSelection(0);
		
		shell.open();

		
		setRowMeta();
		 
		
		while (!shell.isDisposed()) 
		{ 
			if (!display.readAndDispatch())
				display.sleep(); 
		}

		return stepname;

	}   
	
	private String[] getDBMeta(){ 
        
		 return transMeta.getDatabaseNames();
	 
	}
	//初始化数据
	private void initInfo(){ 
		if(input.getOutputName()==null){  
			className_text.setText("");
		}else{
			className_text.setText(input.getOutputName());
		}
		 
		 
		
		XMLField[] nrfields = input.getInputField();  
		
		for (int i = 0; i < nrfields.length; i++) {
			XMLField field = nrfields[i];
			TableItem item = input_view.table.getItem(i); 
            if (field.getFieldName()!=null) item.setText(1, field.getFieldName()); 
            if (field.getFormat()!=null) item.setText(2, field.getFormat());
            item.setText(3, field.getTypeDesc());
            if (field.getLength()>=0) item.setText(4, ""+field.getLength());
            if (field.getPrecision()>=0) item.setText(5, ""+field.getPrecision()); 
		}    
 
		 
		
	}   
	//设置列名
	private String[] setRowMeta() {
		String[] rowNames = null;
		StepMeta stepMeta = transMeta.findStep(stepname);
		if (stepMeta != null) {
			try {
				RowMetaInterface row = transMeta.getPrevStepFields(stepMeta);
				if (row == null) {
					return null;

				}
				rowNames = new String[row.size()]; 
				for (int i = 0; i < row.size(); i++) {
					rowNames[i] = row.getValueMeta(i).getName();
				}
			} catch (KettleException e) {
				logError(BaseMessages.getString(PKG, "System.Dialog.GetFieldsFailed.Message"));
			}
		}
		return rowNames;
	}
		
	private void cancel()

	{  
		dispose();

	} 
	//点击确定按钮操作,并保存数据
	private void ok() 
	{ 
		 
		if (!"".equals(className_text.getText())) 
		{   
			
			input.setOutputName(className_text.getText());
			 
			int infields = input_view.nrNonEmpty();
			input.allocateInput(infields);
			 
	        for (int i=0;i<infields;i++)
	        {
	            XMLField field = new XMLField(); 
	            TableItem item = input_view.getNonEmpty(i);
	            field.setFieldName( item.getText(1) );  
	            field.setFormat( item.getText(2) );
	            field.setType( item.getText(3) );
	           
	            field.setLength( Const.toInt(item.getText(4), -1) );
	            field.setPrecision( Const.toInt(item.getText(5), -1) ); 
	            input.getInputFields()[i]  = field;
	        }  
	    	 
		    dispose();

			return;

		}else{
			   msgError.setMessage("请输入需要解析的对象"); 
		} 
		msgError.setMessage("请输入需要解析的对象");

		msgError.open();
//
	} 
}

5:JSONToString
说明：实现json转string

public boolean processRow(StepMetaInterface smi, StepDataInterface sdi) throws KettleException
	{  
		meta = (JSONToStringMeta) smi;
		data = (JSONToStringData) sdi;
		//获得转换流程中的数据
		Object[] row = getRow();   
		if (row == null)   
		{
			setOutputDone(); 
			return false; 
		} 
		// 初始数据
		if (first) 
		{ 
			first = false; 
			//获得转换流程中的例字段
			data.outputRowMeta = getInputRowMeta().clone(); 
			//往转换流程中新增字段
			meta.getFields(data.outputRowMeta, getStepname(), null, null, this); 
		} 
		if(meta.getOutputName()==null){
			logError("no find class,class name  is ["+meta.getOutputName()+ "]");
			return false;
		}  
		String msg = "Success";
		String code = "200"; 
		Map<String, Object> res = new  HashMap<String, Object>(); 
		res.put(RES_CODE, code);
		res.put(RES_MESSAGE, msg); 
		Map<String,Object> items=getInputRows(  row); 
		//格式转换
		String out=JSON.toJSONString(items);
		//往转换流程中新增数据
		Object[] outData = RowDataUtil.addValueData(row, getInputRowMeta().size(), out); 
		getRes(res, outData, row); 
		return true;
	}