HBaseUtils { public static StructType buildSchema(Map<String, HBaseSerde.ColumnDef> columnDefinitions) { List<StructField> fields = Lists.newArrayList(); for (Map.Entry<String, HBaseSerde.ColumnDef> columnDef : columnDefinitions.entrySet()) { fields.add(DataTypes.createStructField(columnDef.getValue().name, ConfigurationDataTypes.getSparkDataType(columnDef.getValue().type), true)); } return DataTypes.createStructType(fields); }  synchronized static Connection getConnection(Config config); static Configuration getHBaseConfiguration(Config config); static int batchSizeFor(Config config); static TableName tableInfoFor(Config config); static HBaseSerde getSerde(Config config); static byte[] rowKeySeparatorFor(Config config); static List<String> rowKeyFor(Config config); static Map<String, HBaseSerde.ColumnDef> columnsFor(Config config); static StructType buildSchema(Map<String, HBaseSerde.ColumnDef> columnDefinitions); static Scan mergeRangeScans(List<Scan> rangeScans); static byte[] exclusiveStopRow(byte[] startRow); static Validations getValidations(); static final String ZK_QUORUM_PROPERTY; static final String TABLE_NAME_PROPERTY; static final String ROWKEY_PROPERTY; static final String KEY_SEPARATOR; static final String COLUMNS_PROPERTY; static final String SERDE_PROPERTY; static final String HBASE_PASSTHRU_PREFIX; static final String HBASE_BATCH_SIZE; static final int DEFAULT_HBASE_BATCH_SIZE; static final String DEFAULT_SERDE_PROPERTY; static final String DEFAULT_KEY_SEPARATOR; }
@Test public void testBuildSchema() { Config goodConfig = ConfigUtils.configFromResource("/hbase/hbase-output.conf").getConfig("output"); Map<String, HBaseSerde.ColumnDef> columnDefs = HBaseUtils.columnsFor(goodConfig); StructType schema = HBaseUtils.buildSchema(columnDefs); Map<String, DataType> shouldHaveFields = Maps.newHashMap(); shouldHaveFields.put("symbol", DataTypes.StringType); shouldHaveFields.put("transacttime", DataTypes.LongType); shouldHaveFields.put("clordid", DataTypes.StringType); shouldHaveFields.put("orderqty", DataTypes.IntegerType); shouldHaveFields.put("leavesqty", DataTypes.IntegerType); shouldHaveFields.put("cumqty", DataTypes.IntegerType); StructField[] fields = schema.fields(); assertEquals("Schema should contain 6 fields", 6, fields.length); Map<String, DataType> hasFields = Maps.newHashMap(); for (StructField field : fields) { hasFields.put(field.name(), field.dataType()); } for (Map.Entry<String, DataType> shouldHave : shouldHaveFields.entrySet()) { assertTrue("Schema should have field: " + shouldHave.getKey(), hasFields.containsKey(shouldHave.getKey())); assertEquals("Field " + shouldHave.getKey() + " should have type: " + shouldHave.getValue(), shouldHave.getValue(), hasFields.get(shouldHave.getKey())); } }