在 Google DataFlow (java) 中创建复杂的 BigQuery Schema

Question

我想要加载到 BigQuery 中的复杂对象的无限流。这些对象的结构表示我在 BigQuery 中的目标表的架构。

问题在于，由于 POJO 中有很多嵌套字段，因此将它的 extremely tedious task 转换为 TableSchema 对象，我正在寻找一种快速/自动化的方法将我的 POJO 转换为 @写入 BigQuery 时出现 987654323@ 对象。

我对 Apache Beam API 不是很熟悉，如有任何帮助，我们将不胜感激。

Answer 1

在管道中，我从 GCS 加载架构列表。我将它们保留为字符串格式，因为 TableSchema 不可序列化。但是，我将它们加载到 TableSchema 以进行验证。 然后我将它们以字符串格式添加到 Option 对象中的映射中。

String schema = new String(blob.getContent());
// Decorate list of fields for allowing a correct parsing
String targetSchema = "{\"fields\":" + schema + "}";
try {
    //Preload schema to ensure validity, but then use string version
    Transport.getJsonFactory().fromString(targetSchema, TableSchema.class);

    String tableName = blob.getName().replace(SCHEMA_FILE_PREFIX, "").replace(SCHEMA_FILE_SUFFIX, "");
    tableSchemaStringMap.put(tableName, targetSchema);
} catch (IOException e) {
    logger.warn("impossible to read schema " + blob.getName() + " in bucket gs://" + options.getSchemaBucket());
}

我在开发这个时没有找到其他解决方案。

Answer 2

在我的公司中，我创建了一种 ORM（我们称为 OBQM）来执行此操作。我们期待着向公众发布它。代码很大（特别是因为我创建了注释等），但我可以与您分享一些 sn-ps 以快速生成模式：

public TableSchema generateTableSchema(@Nonnull final Class cls) {

        final TableSchema tableSchema = new TableSchema();
        tableSchema.setFields(generateFieldsSchema(cls));

        return tableSchema;
    }

public List<TableFieldSchema> generateFieldsSchema(@Nonnull final Class cls) {

        final List<TableFieldSchema> schemaFields = new ArrayList<>();
        final Field[] clsFields = cls.getFields();

        for (final Field field : clsFields) {
            schemaFields.add(fromFieldToSchemaField(field));
        }

        return schemaFields;
    }

此代码从 POJO 类中获取所有字段并创建一个 TableSchema 对象（BigQueryIO 在 ApacheBeam 中使用的对象）。您可以看到我创建的一个名为fromFieldToSchemaField 的方法。该方法识别每个字段类型并设置字段名称、模式、描述和类型。在这种情况下，为了简单起见，我将重点关注类型和名称：

public static TableFieldSchema fromFieldToSchemaField(@Nonnull final Field field) {
        return fromFieldToSchemaField(field, 0);
}

public static TableFieldSchema fromFieldToSchemaField(
            @Nonnull final Field field,
            final int iteration) {

        final TableFieldSchema schemaField = new TableFieldSchema();
        final Type customType = field.getGenericType().getTypeName()

        schemaField.setName(field.getName());
        schemaField.setMode("NULLABLE"); // You can add better logic here, we use annotations to override this value
        schemaField.setType(getFieldTypeString(field));
        schemaField.setDescription("Optional"); // Optional

        if (iteration < MAX_RECURSION
                && (isStruct(schemaField.getType())
                        || isRecord(schemaField.getType()))) {
            final List<TableFieldSchema> schemaFields = new ArrayList<>();
            final Field[] fields = getFieldsFromComplexObjectField(field);

            for (final Field subField : fields) {
                schemaFields.add(
                        fromFieldToSchemaField(
                                subField, iteration + 1));
            }

            schemaField.setFields(schemaFields.isEmpty() ? null : schemaFields);
        }

        return schemaField;
    }

现在是返回 BigQuery 字段类型的方法。

public static String getFieldTypeString(@Nonnull final Field field) {
   // On my side this code is much complex but this is a short version of that
   final Class<?> cls = (Class<?>) field.getGenericType()
   if (cls.isAssignableFrom(String.class)) {
            return "STRING";
        } else if (cls.isAssignableFrom(Integer.class) || cls.isAssignableFrom(Short.class)) {
            return "INT64";
        } else if (cls.isAssignableFrom(Double.class)) {
            return "NUMERIC";
        } else if (cls.isAssignableFrom(Float.class)) {
            return "FLOAT64";
        } else if (cls.isAssignableFrom(Boolean.class)) {
            return "BOOLEAN";
        } else if (cls.isAssignableFrom(Double.class)) {
            return "BYTES";
        } else if (cls.isAssignableFrom(Date.class)
                || cls.isAssignableFrom(DateTime.class)) {
            return "TIMESTAMP";
        } else {
            return "STRUCT";
        }
    }

请记住，我并没有展示如何识别原始类型或数组。但这对您的代码来说是一个好的开始:)。如果您需要任何帮助，请告诉我。

Answer 3

如果您在 PubSub 中使用 JSON 进行消息序列化，您可以使用提供的模板之一：

PubSub To BigQuery Template

该模板的代码在这里：

PubSubToBigQuery.java