Project

General

Profile

« Previous | Next » 

Revision 48800

Added by Claudio Atzori over 6 years ago

more logging in the oozie job monitor procedure that tries to fetch the job report. More java8 refactorings

View differences:

HadoopServiceCore.java
2 2 

  		




  3
  3
  
    
import java.io.IOException;


  		




  4
  4
  
    
import java.net.URI;


  		




  5
  
  
    
import java.util.List;


  		




  6
  
  
    
import java.util.Map;


  		




  
  5
  
    
import java.util.*;


  		




  7
  6
  
    
import java.util.Map.Entry;


  		




  8
  
  
    
import java.util.NavigableMap;


  		




  9
  
  
    
import java.util.Set;


  		




  
  7
  
    
import java.util.stream.Collectors;


  		




  10
  8
  
    

  		




  11
  
  
    
import com.google.common.base.Function;


  		




  12
  
  
    
import com.google.common.collect.Iterables;


  		




  13
  9
  
    
import com.google.common.collect.Lists;


  		




  14
  10
  
    
import com.google.common.collect.Maps;


  		




  15
  11
  
    
import com.google.common.collect.Sets;


  		




  ......




  20
  16
  
    
import eu.dnetlib.data.hadoop.rmi.hbase.HBaseRowDescriptor;


  		




  21
  17
  
    
import eu.dnetlib.data.hadoop.rmi.hbase.schema.HBaseTableDescriptor;


  		




  22
  18
  
    
import eu.dnetlib.data.hadoop.rmi.hbase.schema.HBaseTableRegionInfo;


  		




  23
  
  
    
import eu.dnetlib.data.hadoop.utils.HadoopUtils;


  		




  24
  19
  
    
import org.apache.commons.lang.StringUtils;


  		




  25
  20
  
    
import org.apache.commons.logging.Log;


  		




  26
  21
  
    
import org.apache.commons.logging.LogFactory;


  		




  ......




  46
  41
  
    

  		




  47
  42
  
    
	public List<String> listTables(final ClusterName clusterName) throws IOException, HadoopServiceException {


  		




  48
  43
  
    
		final HBaseAdmin admin = getHBaseAdmin(clusterName);


  		




  49
  
  
    
		return Lists.newArrayList(Iterables.transform(Lists.newArrayList(admin.listTables()), HadoopUtils.tableName()));


  		




  
  44
  
    
		return Arrays.asList(admin.listTables())


  		




  
  45
  
    
				.stream()


  		




  
  46
  
    
				.map(HTableDescriptor::getNameAsString)


  		




  
  47
  
    
				.collect(Collectors.toList());


  		




  50
  48
  
    
	}


  		




  51
  49
  
    

  		




  52
  50
  
    
	private HBaseAdmin getHBaseAdmin(final ClusterName clusterName) throws HadoopServiceException {


  		




  ......




  70
  68
  
    

  		




  71
  69
  
    
		final Set<String> columns = Sets.newHashSet();


  		




  72
  70
  
    

  		




  73
  
  
    
		for (HColumnDescriptor hColDesc : Lists.newArrayList(desc.getColumnFamilies())) {


  		




  
  71
  
    
		for (HColumnDescriptor hColDesc : Arrays.asList(desc.getColumnFamilies())) {


  		




  74
  72
  
    
			columns.add(hColDesc.getNameAsString());


  		




  75
  73
  
    
		}


  		




  76
  74
  
    

  		




  ......




  94
  92
  
    
	public List<String> describeTable(final ClusterName clusterName, final String table) throws IOException, HadoopServiceException {


  		




  95
  93
  
    
		final HBaseAdmin admin = getHBaseAdmin(clusterName);


  		




  96
  94
  
    
		final HTableDescriptor desc = admin.getTableDescriptor(table.getBytes());


  		




  97
  
  
    

  		




  98
  
  
    
		return Lists.newArrayList(Iterables.transform(desc.getFamilies(), new Function<HColumnDescriptor, String>() {


  		




  99
  
  
    

  		




  100
  
  
    
			@Override


  		




  101
  
  
    
			public String apply(final HColumnDescriptor desc) {


  		




  102
  
  
    
				return desc.getNameAsString();


  		




  103
  
  
    
			}


  		




  104
  
  
    
		}));


  		




  
  95
  
    
		return desc.getFamilies().stream()


  		




  
  96
  
    
				.map(d -> d.getNameAsString())


  		




  
  97
  
    
				.collect(Collectors.toList());


  		




  105
  98
  
    
	}


  		




  106
  99
  
    

  		




  107
  100
  
    
	public void truncateTable(final ClusterName clusterName, final String table) throws IOException, HadoopServiceException {


  		




  ......




  201
  194
  
    
			createTable(clusterName, table, columns);


  		




  202
  195
  
    
		} else {


  		




  203
  196
  
    
			final HTableDescriptor desc = admin.getTableDescriptor(Bytes.toBytes(table));


  		




  204
  
  
    
			final Set<String> foundColumns = Sets.newHashSet(Iterables.transform(Lists.newArrayList(desc.getColumnFamilies()), HadoopUtils.columnName()));


  		




  205
  197
  
    

  		




  
  198
  
    
			final Set<String> foundColumns = desc.getFamilies().stream()


  		




  
  199
  
    
					.map(d -> d.getNameAsString())


  		




  
  200
  
    
					.collect(Collectors.toCollection(HashSet::new));


  		




  
  201
  
    

  		




  206
  202
  
    
			log.info("ensuring columns on table " + table + ": " + columns);


  		




  207
  
  
    
			final List<String> missingColumns = Lists.newArrayList(Iterables.filter(columns, HadoopUtils.columnPredicate(foundColumns)));


  		




  
  203
  
    
			final Collection<String> missingColumns = Sets.difference(columns, foundColumns);


  		




  208
  204
  
    
			if (!missingColumns.isEmpty()) {


  		




  209
  205
  
    

  		




  210
  206
  
    
				if (admin.isTableEnabled(table)) {


  		












Also available in:  Unified diff