ParquetOutput extends BaseStep implements StepInterface { public void init( RowMetaInterface rowMeta ) throws Exception { FormatService formatService; try { formatService = meta.getNamedClusterResolver().getNamedClusterServiceLocator() .getService( getNamedCluster(), FormatService.class ); } catch ( ClusterInitializationException e ) { throw new KettleException( "can't get service format shim ", e ); } if ( meta.getFilename() == null ) { throw new KettleException( "No output files defined" ); } data.output = formatService.createOutputFormat( IPentahoParquetOutputFormat.class, getNamedCluster() ); String outputFileName = environmentSubstitute( meta.constructOutputFilename() ); pvfsFileAliaser = new PvfsFileAliaser( outputFileName, getTransMeta(), data.output, meta.overrideOutput, getLogChannel() ); data.output.setOutputFile( pvfsFileAliaser.generateAlias(), meta.overrideOutput ); data.output.setFields( meta.getOutputFields() ); IPentahoParquetOutputFormat.COMPRESSION compression; try { compression = IPentahoParquetOutputFormat.COMPRESSION.valueOf( meta.getCompressionType( variables ).name().toUpperCase() ); } catch ( Exception ex ) { compression = IPentahoParquetOutputFormat.COMPRESSION.UNCOMPRESSED; } data.output.setCompression( compression ); data.output .setVersion( ParquetOutputMetaBase.ParquetVersion.PARQUET_1.equals( meta.getParquetVersion( variables ) ) ? IPentahoParquetOutputFormat.VERSION.VERSION_1_0 : IPentahoParquetOutputFormat.VERSION.VERSION_2_0 ); if ( meta.getRowGroupSize( variables ) > 0 ) { data.output.setRowGroupSize( meta.getRowGroupSize( variables ) * 1024 * 1024 ); } if ( meta.getDataPageSize( variables ) > 0 ) { data.output.setDataPageSize( meta.getDataPageSize( variables ) * 1024 ); } data.output.enableDictionary( meta.enableDictionary ); if ( meta.getDictPageSize( variables ) > 0 ) { data.output.setDictionaryPageSize( meta.getDictPageSize( variables ) * 1024 ); } data.writer = data.output.createRecordWriter(); } ParquetOutput( StepMeta stepMeta, StepDataInterface stepDataInterface, int copyNr, TransMeta transMeta,
                        Trans trans ); @Override synchronized boolean processRow( StepMetaInterface smi, StepDataInterface sdi ); void init( RowMetaInterface rowMeta ); void closeWriter(); boolean init( StepMetaInterface smi, StepDataInterface sdi );  }
@Test public void initShouldPassEmbeddedMetastoreKey() { ParquetOutputMeta stepMetaInterface = mock( ParquetOutputMeta.class ); ParquetOutputData stepDataInterface = mock( ParquetOutputData.class ); NamedClusterEmbedManager namedClusterEmbedManager = mock( NamedClusterEmbedManager.class ); when( mockTransMeta.getNamedClusterEmbedManager() ).thenReturn( namedClusterEmbedManager ); when( mockTransMeta.getEmbeddedMetastoreProviderKey() ).thenReturn( "metastoreProviderKey" ); parquetOutput.init( stepMetaInterface, stepDataInterface ); verify( namedClusterEmbedManager ).passEmbeddedMetastoreKey( mockTransMeta, "metastoreProviderKey" ); }