@Udf
  public Map<String, String> records(@UdfParameter final String jsonObj) {
    if (jsonObj == null) {
      return null;
    }

    final JsonNode node = UdfJsonMapper.parseJson(jsonObj);
    if (node.isMissingNode() || !node.isObject()) {
      return null;
    }

    final Map<String, String> ret = new HashMap<>(node.size());
    node.fieldNames().forEachRemaining(k -> {
      final JsonNode value = node.get(k);
      if (value instanceof TextNode) {
        ret.put(k, value.textValue());
      } else {
        ret.put(k, value.toString());
      }
    });
    return ret;
  }
@Test
  public void shouldExtractRecords() {
    // When
    final Map<String, String> result = udf.records("{\"a\": \"abc\", \"b\": { \"c\": \"a\" }, \"d\": 1}");

    // Then:
    final Map<String, String> expected = new HashMap<String, String>() {{
        put("a", "abc");
        put("b", "{\"c\":\"a\"}");
        put("d", "1");
      }};

    assertEquals(expected, result);
  }