@Test public void requireThatUDFGeneratesComplexDataTypes() throws Exception { Schema schema = new Schema(); Tuple tuple = TupleFactory.getInstance().newTuple(); Tuple intTuple = TupleFactory.getInstance().newTuple(); int[] intArray = {1, 2, 3}; for (int i : intArray) { intTuple.append(i); } Tuple stringTuple = TupleFactory.getInstance().newTuple(); String[] stringArray = {"a", "b", "c"}; for (String s : stringArray) { stringTuple.append(s); } DataBag bag = new SortedDataBag(null); bag.add(intTuple); bag.add(stringTuple); Map<String, Object> innerMap = new HashMap<String, Object>() {{ put("a", "string"); put("tuple", intTuple); }}; DataByteArray bytes = new DataByteArray("testdata".getBytes()); Map<String, Object> outerMap = new HashMap<String, Object>() {{ put("string", "value"); put("int", 3); put("float", 3.145); put("bool", true); put("byte", bytes); put("map", innerMap); put("bag", bag); }}; addToTuple("map", DataType.MAP, outerMap, schema, tuple); VespaDocumentOperation docOp = new VespaDocumentOperation("docid=empty"); docOp.setInputSchema(schema); String json = docOp.exec(tuple); ObjectMapper m = new ObjectMapper(); JsonNode root = m.readTree(json); JsonNode fields = root.get("fields"); JsonNode map = fields.get("map"); assertEquals("value", map.get("string").getTextValue()); assertEquals(3, map.get("int").getIntValue()); assertEquals(3.145, map.get("float").getDoubleValue(), 1e-6); assertTrue(map.get("bool").getBooleanValue()); assertEquals("dGVzdGRhdGE=", map.get("byte").getTextValue()); assertEquals("string", map.get("map").get("a").getTextValue()); for (int i = 0; i < intArray.length; ++i) { assertEquals(intArray[i], map.get("map").get("tuple").get(i).asInt()); } JsonNode bagField = map.get("bag"); for (int i = 0; i < intArray.length; ++i) { assertEquals(intArray[i], bagField.get(0).get(i).asInt()); } for (int i = 0; i < stringArray.length; ++i) { assertEquals(stringArray[i], bagField.get(1).get(i).asText()); } }
@Override public String exec(Tuple tuple) throws IOException { if (tuple == null || tuple.size() == 0) { if (statusReporter != null) { statusReporter.incrCounter("Vespa Document Operation Counters", "Document operation failed", 1); } return null; } if (template == null || template.length() == 0) { if (statusReporter != null) { statusReporter.incrCounter("Vespa Document Operation Counters", "Document operation failed", 1); } warnLog("No valid document id template found. Skipping.", PigWarning.UDF_WARNING_1); return null; } if (operation == null) { warnLog("No valid operation found. Skipping.", PigWarning.UDF_WARNING_2); return null; } String json = null; try { if (reporter != null) { reporter.progress(); } Schema inputSchema = getInputSchema(); Map<String, Object> fields = TupleTools.tupleMap(inputSchema, tuple); String docId = TupleTools.toString(fields, template); if (verbose) { System.out.println("Processing docId: "+ docId); } json = create(operation, docId, fields, properties, inputSchema); if (json == null || json.length() == 0) { warnLog("No valid document operation could be created.", PigWarning.UDF_WARNING_3); return null; } } catch (Exception e) { if (statusReporter != null) { statusReporter.incrCounter("Vespa Document Operation Counters", "Document operation failed", 1); } StringBuilder sb = new StringBuilder(); sb.append("Caught exception processing input row: \n"); sb.append(tuple.toString()); sb.append("\nException: "); sb.append(getStackTraceAsString(e)); warnLog(sb.toString(), PigWarning.UDF_WARNING_4); return null; } if (statusReporter != null) { statusReporter.incrCounter("Vespa Document Operation Counters", "Document operation ok", 1); } return json; }
VespaDocumentOperation extends EvalFunc<String> { @Override public String exec(Tuple tuple) throws IOException { if (tuple == null || tuple.size() == 0) { if (statusReporter != null) { statusReporter.incrCounter("Vespa Document Operation Counters", "Document operation failed", 1); } return null; } if (template == null || template.length() == 0) { if (statusReporter != null) { statusReporter.incrCounter("Vespa Document Operation Counters", "Document operation failed", 1); } warnLog("No valid document id template found. Skipping.", PigWarning.UDF_WARNING_1); return null; } if (operation == null) { warnLog("No valid operation found. Skipping.", PigWarning.UDF_WARNING_2); return null; } String json = null; try { if (reporter != null) { reporter.progress(); } Schema inputSchema = getInputSchema(); Map<String, Object> fields = TupleTools.tupleMap(inputSchema, tuple); String docId = TupleTools.toString(fields, template); if (verbose) { System.out.println("Processing docId: "+ docId); } json = create(operation, docId, fields, properties, inputSchema); if (json == null || json.length() == 0) { warnLog("No valid document operation could be created.", PigWarning.UDF_WARNING_3); return null; } } catch (Exception e) { if (statusReporter != null) { statusReporter.incrCounter("Vespa Document Operation Counters", "Document operation failed", 1); } StringBuilder sb = new StringBuilder(); sb.append("Caught exception processing input row: \n"); sb.append(tuple.toString()); sb.append("\nException: "); sb.append(getStackTraceAsString(e)); warnLog(sb.toString(), PigWarning.UDF_WARNING_4); return null; } if (statusReporter != null) { statusReporter.incrCounter("Vespa Document Operation Counters", "Document operation ok", 1); } return json; } }
VespaDocumentOperation extends EvalFunc<String> { @Override public String exec(Tuple tuple) throws IOException { if (tuple == null || tuple.size() == 0) { if (statusReporter != null) { statusReporter.incrCounter("Vespa Document Operation Counters", "Document operation failed", 1); } return null; } if (template == null || template.length() == 0) { if (statusReporter != null) { statusReporter.incrCounter("Vespa Document Operation Counters", "Document operation failed", 1); } warnLog("No valid document id template found. Skipping.", PigWarning.UDF_WARNING_1); return null; } if (operation == null) { warnLog("No valid operation found. Skipping.", PigWarning.UDF_WARNING_2); return null; } String json = null; try { if (reporter != null) { reporter.progress(); } Schema inputSchema = getInputSchema(); Map<String, Object> fields = TupleTools.tupleMap(inputSchema, tuple); String docId = TupleTools.toString(fields, template); if (verbose) { System.out.println("Processing docId: "+ docId); } json = create(operation, docId, fields, properties, inputSchema); if (json == null || json.length() == 0) { warnLog("No valid document operation could be created.", PigWarning.UDF_WARNING_3); return null; } } catch (Exception e) { if (statusReporter != null) { statusReporter.incrCounter("Vespa Document Operation Counters", "Document operation failed", 1); } StringBuilder sb = new StringBuilder(); sb.append("Caught exception processing input row: \n"); sb.append(tuple.toString()); sb.append("\nException: "); sb.append(getStackTraceAsString(e)); warnLog(sb.toString(), PigWarning.UDF_WARNING_4); return null; } if (statusReporter != null) { statusReporter.incrCounter("Vespa Document Operation Counters", "Document operation ok", 1); } return json; } VespaDocumentOperation(String... params); }
VespaDocumentOperation extends EvalFunc<String> { @Override public String exec(Tuple tuple) throws IOException { if (tuple == null || tuple.size() == 0) { if (statusReporter != null) { statusReporter.incrCounter("Vespa Document Operation Counters", "Document operation failed", 1); } return null; } if (template == null || template.length() == 0) { if (statusReporter != null) { statusReporter.incrCounter("Vespa Document Operation Counters", "Document operation failed", 1); } warnLog("No valid document id template found. Skipping.", PigWarning.UDF_WARNING_1); return null; } if (operation == null) { warnLog("No valid operation found. Skipping.", PigWarning.UDF_WARNING_2); return null; } String json = null; try { if (reporter != null) { reporter.progress(); } Schema inputSchema = getInputSchema(); Map<String, Object> fields = TupleTools.tupleMap(inputSchema, tuple); String docId = TupleTools.toString(fields, template); if (verbose) { System.out.println("Processing docId: "+ docId); } json = create(operation, docId, fields, properties, inputSchema); if (json == null || json.length() == 0) { warnLog("No valid document operation could be created.", PigWarning.UDF_WARNING_3); return null; } } catch (Exception e) { if (statusReporter != null) { statusReporter.incrCounter("Vespa Document Operation Counters", "Document operation failed", 1); } StringBuilder sb = new StringBuilder(); sb.append("Caught exception processing input row: \n"); sb.append(tuple.toString()); sb.append("\nException: "); sb.append(getStackTraceAsString(e)); warnLog(sb.toString(), PigWarning.UDF_WARNING_4); return null; } if (statusReporter != null) { statusReporter.incrCounter("Vespa Document Operation Counters", "Document operation ok", 1); } return json; } VespaDocumentOperation(String... params); @Override String exec(Tuple tuple); static String create(Operation op, String docId, Map<String, Object> fields, Properties properties,
                                Schema schema); }
VespaDocumentOperation extends EvalFunc<String> { @Override public String exec(Tuple tuple) throws IOException { if (tuple == null || tuple.size() == 0) { if (statusReporter != null) { statusReporter.incrCounter("Vespa Document Operation Counters", "Document operation failed", 1); } return null; } if (template == null || template.length() == 0) { if (statusReporter != null) { statusReporter.incrCounter("Vespa Document Operation Counters", "Document operation failed", 1); } warnLog("No valid document id template found. Skipping.", PigWarning.UDF_WARNING_1); return null; } if (operation == null) { warnLog("No valid operation found. Skipping.", PigWarning.UDF_WARNING_2); return null; } String json = null; try { if (reporter != null) { reporter.progress(); } Schema inputSchema = getInputSchema(); Map<String, Object> fields = TupleTools.tupleMap(inputSchema, tuple); String docId = TupleTools.toString(fields, template); if (verbose) { System.out.println("Processing docId: "+ docId); } json = create(operation, docId, fields, properties, inputSchema); if (json == null || json.length() == 0) { warnLog("No valid document operation could be created.", PigWarning.UDF_WARNING_3); return null; } } catch (Exception e) { if (statusReporter != null) { statusReporter.incrCounter("Vespa Document Operation Counters", "Document operation failed", 1); } StringBuilder sb = new StringBuilder(); sb.append("Caught exception processing input row: \n"); sb.append(tuple.toString()); sb.append("\nException: "); sb.append(getStackTraceAsString(e)); warnLog(sb.toString(), PigWarning.UDF_WARNING_4); return null; } if (statusReporter != null) { statusReporter.incrCounter("Vespa Document Operation Counters", "Document operation ok", 1); } return json; } VespaDocumentOperation(String... params); @Override String exec(Tuple tuple); static String create(Operation op, String docId, Map<String, Object> fields, Properties properties,
                                Schema schema);  }