Универсальный Солдат: Groovy Transformer в DataStage

Возможности ETL средства IBM DataStage покрывают достаточно широкий спектр требований, которые возникают в задачах по интеграции данных, но, рано или поздно, возникает потребность расширить функциональные возможности, внедряя Parallel Routines на языке С или создавая Java классы, которые, в дальнейшем используются в Java Transformer или Java Client. Довольно ограниченные возможности же встроенного языка Basic давно устарели и не могут расцениваться как серьезное подспорье (так, например, невозможно использовать XML структуры, или, другой пример — попробуйте написать хеширование MD5 при помощи Basic. Это возможно, но займет значительное время на разработку и отладку).
Как бы там ни было, хотелось бы иметь достаточно гибкое средство, позволяющее работать с потоком данных, не требующее перекомпиляции своих исходных кодов и которое можно было бы использовать в редакторе DataStage Client. Моим коллегой и близким другом было предложено разработать Groovy Transformer. О нем и пойдет речь в данной заметке.

Почему Groovy? Потому что этот язык достаточно гибок и обладает всеми возможностями Java, так как является надстройкой над этим языком, но помимо этого он предлагает разработчикам следующие преимущества:

• Нативный синтакс для хешей (ассоциативных массивов) и списков, так, вместо следующего кода на Java

  import java.util.* ;
  …
  HashMap<String,String> someMap=new HashMap<String,String>() ;
  someMap.put("Key1", "Value") ;
  …
  String valFromMap=someMap.get("Key1") ;
  
  

  можно использовать следующий код Groovy (java.util.* в нем уже импортирована по умолчанию):

  HashMap someMap=new HashMap();
  someMap.Key1= "Value" ;
  …
  valFromMap=someMap.Key1;
  

• Интерпретация строк (почему все ее называют интерполяцией?):

  def NumberOfItems= 10;
  println "Number of items=$NumberOfItems";
  

• Динамическое выполнение кода, можно выполнить код, хранимый в строковой переменной, в потоке или в файлах без компиляции:

  def GroovyCode='sum=a+b; println "Sum is $sum"';
  Binding binding=new Binding();
  binding.setVariable("a", 10);
  binding.setVariable("b", 20);
  GroovyShell shell = new GroovyShell(binding);
  shell.evaluate(GroovyCode); 
  

• Нативная поддержка XML и JSON:

  def writer=new StringWriter() ;
  def xmlOut=new groovy.xml.MarkupBuilder(writer) ;
  xmlOut.JobsInfo{
      Job(name : 'ODS_MOUVEMENT_C') {
        precedants {
           precedant "ODS_ECRITURE_C"
           precedant "ODS_TEC_DEB"
        }
   }
  }
  println writer;
  

  Этот пример должен распечатать следующий XML:

  <JobsInfo>
    <Job name='ODS_MOUVEMENT_C'>
      <precedants>
        <precedant>ODS_ECRITURE_C</precedant>
        <precedant>ODS_TEC_DEB</precedant>
      </precedants>
    </Job>
  </JobsInfo>
  

  если вы хотите сгенерировать JSON, нужно заменить groovy.xml.MarkupBuilder на groovy.json.JsonBuilder.

Основная идея Groovy Transformer заключается в использовании кода на языке Groovy в Java Transformer. Поскольку код можно писать прямо в этом стейдже, то вы можете сами решать, как вам поступать – выполнять Groovy-код, который хранится в файле, приходит из параметров джоба или который вы напишете сами.
Таким образом, нам надо научиться создавать Java Transformer. Те, кто уже знают, как это реализуется, может пропустить этот раздел. Но я постараюсь кратко, так как документация по этой части написана достаточно подробно.
Итак, чтобы создать Ява-трансформер, нам нужно создать класс, который наследуется от класса Stage:

import com.ascentialsoftware.jds.* ;

class MyJavaTransformer extends Stage{

}

И необходимо имплементировать три самых часто используемых метода: initialize(), process() и terminate().
Метод initialize() выполняется перед обработкой стейджем строк потока и может содержать объявления объектов, которые вы собираетесь использовать в течении всей жизни трансформера.
Метод process() выполняется для каждой линии входного потока и должен содержать логику вашей обработки.
Метод terminate() выполняется в конце существования трансформера и в нем могут содержаться действия для удаления временных объектов (да знаю я, что нет деструкторов в Яве, имеется в виду любой мусор, который вы использовали: файлы, таблицы, мало ли что).
Замечание для параллельного режима трансформера: DataStage запускает отдельную Яву-машину на каждую ноду. Другими словамИ, если у вас четыре ноды, то DataStage запустит четыре JVM. Так как виртуальные машины изолированы, то у вас нет приемлемых способов обмена данными, между потоками, запущенными в каждой из них.

Теперь мы готовы к созданию шаблону нашего Ява-трансформера:

import com.ascentialsoftware.jds.*;

public class MyJavaTransformer extends Stage {

    public void initialize() {
        trace("Init");
    }

    public void terminate() {
        trace("Terminate");
    }

    public int process() {
         return 0;
    }
}

Для чтения входящих в трансформер строк и выходящих из него, можно использовать объект Row и два метода: readRow() для доступа к значениям входного потока и writeRow() для записи в выходной.
Объекь Row также дает возможность получить метаданные каждой колонки и позволяет получить значения этих колонок. Следующий пример демонстрирует, каким образом можно заменить значения всех колонок, которые имеют тип VarChar значением « Hello from the Java », все остальные колонки «проталкиваются» без изменений дальше:

public int process() {

Row inputRow=readRow() ;
if (inputRow == null) 
	//нет больше строк в потоке
    return OUTPUT_STATUS_END_OF_DATA;
Row outputRow=createOutputRow();

for (int i=0;i<inputRow.getColumnCount();i++) {
	Object column=inputRow.getValueAsSQLTyped(i);
	if (column instanceof java.lang.String) 
		outputRow.setValueAsSQLTyped(i, “Hello from Java”);
	else
		outputRow.setValueAsSQLTyped(i, column);
}
writeRow(outputRow);
}

Заметка: Чтобы скомпилировать класс трансформера, не забудьте импортировать библиотеку tr4j.jar в class path или в вашей IDE.
Теперь мы можем сформулировать требования к нашему Groovy Transformer.

Groovy Transformer- это JavaTransformer который компилирует на лету код Groovy. В него добавлен синтаксический сахар для облегчения рутинных операций, которые приходится выполнять при работе с классом Stage

Итак, наш трансформер должен:

• Получать код Groovy из вкладки Stage->Properties стейджа Java Transformer ;
• Позволять получать доступ к метаданным входного и выходного потоков;
• Позволять манипулировать данными колонок по их имени (вместо номера колонки).

В соответствии с этим требованиями, был создан Groovy Transformer, который вы можете скачать здесь: http://geckelberryfinn.ru/fr/GroovyTransformer.html ^[1]. (Осторожно! Этот Java Transformer также написан на Groovy =), с декомпиляцией будут проблемы).

Groovy Transformer предопределяет следующие объекты:

Таким образом, есть два способа создания выходного потока:

1. Заполнить список OutputRows ;
2. Вызвать метод createOutputRow(), а затем — writeRow объекта GTransformer.

Какой из способов использовать – зависит от конкретной ситуации.

Чтобы начать использовать Groovy Transformer в ваших проектах, неплохо бы заполнить свойства стейджа Java Transformer:

• Transformer Class Name: groovytransformer.GroovyTransformer
• User’s Classpath: /path/to/jar/GroovyTransformer.jar

Ниже я приведу несколько примеров использования Groovy Transformer:

• Легкое извлечение данных из JDBC — источника.
  В этом примере мы извлечем данные из таблицы ODS.AXE_LOCAL используя Oracle как СУБД, хотя для него есть отдельный стейдж. Но, на самом деле, этот код можно использовать в случаях, когда необходимо получать данные из «экзотических источников данных» (Interbase, FoxPro, VectorWise, Derby, SQLite, H2 etc.).
  Этот Groovy-код достаточно универсальный, так как не использует специфическую информацию, кроме как запроса SQL. Иными словами, его можно использовать в любых случаях.
  Для того, чтобы запустить этот пример, не забудьте добавить во вкладке « Output » трансформера список колонок именами, чувствительными к регистру.

  sql = Sql.newInstance( 'jdbc:oracle:thin:@oraclehost:1533/nomdebdd', 
  'user', 'password', 'oracle.jdbc.OracleDriver' );
  
  sql.eachRow('select id, LIBL_AXE_CENTRAL, SYS_CREATED_DAT as DAT_DEB_VALID  from ODS.ODS_AXE_CENTRAL') {  it ->
     GRow=GTransformer.createOutputRow();
     i=0;
     it.toRowResult().each{k,v->
              i++;
              if (v instanceof java.sql.Date)
                 GRow.setValueAsSQLTyped(OutputMatching.get(k),it.getTimestamp(i));
              else 
                 GRow.setValueAsSQLTyped(OutputMatching.get(k),v)  ;
     }
     GTransformer.writeRow(GRow);
  }
  

• Использование регулярных выражений.
  В этом примере мы попытаемся извлечь информацию о контракте (его номер и дату) из строк вида:

  « Контракт F-U-1009 от 10/01/1960 »
  « Договор об оказании услуг F-U-1019 за 10/01/1961 »
  « Кон-т номер F-U-1001 от 10/01/1962 »

  и нам желательно получить на выходе две колонки на выходе: номер контракта и дату его заключения.
  В этом случае можно использовать следующие регулярные выражения:

  • Для номера контракта: [A-Z]+(-[A-Z])+(-[0-9]+)
  • Для его даты: [0-3]?[0-9](/|\.)[0-1]?[0-9](/|\.)(19|20)?[0-9]{2}

  Тогда для этого примера код Groovy может быть таким (обратите внимание, что используется другой способ генерации выходного потока):

  String StrIn=InputColumns.contrat 
  contRegExp="[A-Z]+(-[A-Z])+(-[0-9]+)"
  datRegExp="[0-3]?[0-9](/|\.)[0-1]?[0-9](/|\.)(19|20)?[0-9]{2}"
  
  def matchRegEx(str, regExp) {
  matcher=(str=~regExp);
  if (matcher.getCount()!=0) 
  return matcher[0][0];
  else
  return null;
  }
  
  def cont=matchRegEx(StrIn, contRegExp);
  def dat =matchRegEx(StrIn, datRegExp);
  
  HashMap curRow=new HashMap();
  OutputRows[0]=curRow;
  OutputRows[0].contrat=StrIn;
  OutputRows[0].numero=cont;
  OutputRows[0].date=dat;
  

Несколько полезных ссылок:

1. Справочник Groovy: http://groovy.codehaus.org/Documentation ^[2]
2. Пример Java Transformer : http://www.ibm.com/developerworks/data/library/techarticle/dm-1106etljob/index.html ^[3]

Автор: Geckelberryfinn

Источник ^[4]