private static RuntimeConverter createRuntimeConverter(TypeInformation<?> info) {
		if (info.equals(Types.VOID)) {
			return (csvMapper, container, obj) -> container.nullNode();
		} else if (info.equals(Types.STRING)) {
			return (csvMapper, container, obj) -> container.textNode((String) obj);
		} else if (info.equals(Types.BOOLEAN)) {
			return (csvMapper, container, obj) -> container.booleanNode((Boolean) obj);
		} else if (info.equals(Types.BYTE)) {
			return (csvMapper, container, obj) -> container.numberNode((Byte) obj);
		} else if (info.equals(Types.SHORT)) {
			return (csvMapper, container, obj) -> container.numberNode((Short) obj);
		} else if (info.equals(Types.INT)) {
			return (csvMapper, container, obj) -> container.numberNode((Integer) obj);
		} else if (info.equals(Types.LONG)) {
			return (csvMapper, container, obj) -> container.numberNode((Long) obj);
		} else if (info.equals(Types.FLOAT)) {
			return (csvMapper, container, obj) -> container.numberNode((Float) obj);
		} else if (info.equals(Types.DOUBLE)) {
			return (csvMapper, container, obj) -> container.numberNode((Double) obj);
		} else if (info.equals(Types.BIG_DEC)) {
			return (csvMapper, container, obj) -> container.numberNode((BigDecimal) obj);
		} else if (info.equals(Types.BIG_INT)) {
			return (csvMapper, container, obj) -> container.numberNode((BigInteger) obj);
		} else if (info.equals(Types.SQL_DATE)) {
			return (csvMapper, container, obj) -> container.textNode(obj.toString());
		} else if (info.equals(Types.SQL_TIME)) {
			return (csvMapper, container, obj) -> container.textNode(obj.toString());
		} else if (info.equals(Types.SQL_TIMESTAMP)) {
			return (csvMapper, container, obj) -> container.textNode(obj.toString());
		} else if (info instanceof RowTypeInfo){
			return createRowRuntimeConverter((RowTypeInfo) info, false);
		} else if (info instanceof BasicArrayTypeInfo) {
			return createObjectArrayRuntimeConverter(((BasicArrayTypeInfo) info).getComponentInfo());
		} else if (info instanceof ObjectArrayTypeInfo) {
			return createObjectArrayRuntimeConverter(((ObjectArrayTypeInfo) info).getComponentInfo());
		} else if (info instanceof PrimitiveArrayTypeInfo &&
				((PrimitiveArrayTypeInfo) info).getComponentType() == Types.BYTE) {
			return createByteArrayRuntimeConverter();
		}
		else {
			throw new RuntimeException("Unsupported type information '" + info + "'.");
		}
	}
Creates a {@link CsvRowSerializationSchema} expecting the given {@link TypeInformation}.

@param typeInfo type information used to create schema.
kafka-base/kafka-base-sink/src/main/java/com/dtstack/flink/sql/sink/kafka/serialization/CsvTupleSerializationSchema.java