public static void main(String[] args) {
        PipelineOptions options = PipelineOptionsFactory.create();
        Pipeline pipeline = Pipeline.create(options);
        final List<String> elements = Arrays.asList(
                "1234, Foo",
                "1111, Bar",
                "3456, Baz"

        );

        List<TableFieldSchema> fields = new ArrayList<>();
        fields.add(new TableFieldSchema().setName("id").setType("INTEGER"));
        fields.add(new TableFieldSchema().setName("name").setType("STRING"));
        fields.add(new TableFieldSchema()
                .setName("phone_numbers")
                .setMode("REPEATED")
                .setType("RECORD").setFields(ImmutableList.of(
                        new TableFieldSchema().setName("type").setType("STRING").setMode("REQUIRED"), // remove required mode - BQ IO will insert empty records
                        new TableFieldSchema().setName("number").setType("STRING"))));
        TableSchema schema = new TableSchema().setFields(fields);

        Pipeline p = Pipeline.create(options);

        @UnknownKeyFor
        @NonNull
        @Initialized
        WriteResult apply = p
                // Add a source that uses the hard-coded records.
                .apply(Create.of(elements))
                // Add a transform that converts the records (comma-separated strings) to
                // TableRow.
                .apply("to TableRow", ParDo.of(new DoFn<String, TableRow>() {
                    @ProcessElement
                    public void processElement(ProcessContext c) {
                        String[] columns = c.element().split(", ");

                        TableRow row = new TableRow();
                        row.set("id", Integer.parseInt(columns[0]));
                        row.set("name", columns[1]);
                        row.set("phone_numbers",
                                ImmutableList.of(new TableCell().set("type", "home").set("number", "12345"),
                                        new TableCell().set("type", "work").set("number", "67890")));
                        c.output(row);
                    }
                }))
                .apply("bq", BigQueryIO.writeTableRows()
                        .withSchema(schema)
                        .to("<project id>:<dataset>.repeated2")
                        .withMethod(BigQueryIO.Write.Method.STORAGE_API_AT_LEAST_ONCE)
                        .withCreateDisposition(BigQueryIO.Write.CreateDisposition.CREATE_IF_NEEDED)
                        .withWriteDisposition(BigQueryIO.Write.WriteDisposition.WRITE_APPEND)
                        .withAutoSchemaUpdate(true).ignoreUnknownValues());

        apply.getFailedStorageApiInserts().apply("log", ParDo.of(new LogOutput<>("failed: ")));

        p.run();
    }

    static class LogOutput<T> extends DoFn<T, T> {
        private static final Logger LOG = LoggerFactory.getLogger(LogOutput.class);
        private final String prefix;

        public LogOutput(String prefix) {
            this.prefix = prefix;
        }

        @ProcessElement
        public void processElement(ProcessContext c) throws Exception {
            LOG.info(prefix + c.element());
            c.output(c.element());
        }
    }

BigQueryStorageApiInsertError{row=GenericData{classInfo=[f], {id=1111, name=Bar, phone_numbers=[{type=home, number=12345}, {type=work, number=67890}, {}, {}]}}, errorMessage='Field value of type cannot be empty. on field phone_numbers.'}