代码之家  ›  专栏  ›  技术社区  ›  Mason Wheeler

为什么Snowflake在解析此Avro文件时遇到问题?

  •  0
  • Mason Wheeler  · 技术社区  · 2 年前

    我正在尝试使用Avro文件将数据大容量加载到Snowflake中。我可以创建文件并将其放入Snowflake阶段,但当我运行命令时

    copy into MyTable from '@%"MyTable"/LinkTypes.avro' file_format = (type = avro) match_by_column_name = case_insensitive;
    

    我错了:

    100084(22P02):分析AVRO时出错:不是AVRO文件(魔术不好)

    在十六进制编辑器中查看这个文件,我没有注意到它有什么特别糟糕的地方( Microsoft.Avro.Core 来自NuGet),并且它加载得很好 this random online Avro viewer 我在谷歌上搜索了一下。但不管出于什么原因,雪花都看不懂。

    The file can be found here. 我的Avro数据生成例程如下。有人知道它出了什么问题以及如何修复吗?

    public void Write(string name, IDataReader data, DataDictionary dict)
    {
        using var stream = OpenFileStream(name); // returns a FileStream; assume this works
        var schema = BuildAvroSchema(dict.GetSchema(name)); // returns a JSON string describing the schema; assume this works
        using var writer = AvroContainer.CreateGenericWriter(schema, stream, Codec.Deflate);
        var serializer = AvroSerializer.CreateGeneric(schema);
        WriteReader(reader, writer, stream, serializer.WriterSchema);
    }
    
    private static void WriteReader(IDataReader reader, IAvroWriter<object> writer, Stream stream, TypeSchema schema)
    {
        using var sw = new SequentialWriter<object>(writer, 1024);
        var rec = new AvroRecord(schema);
        var len = reader.FieldCount;
        var buffer = new object[len];
        var names = Enumerable.Range(0, len).Select(i => reader.GetName(i)).ToArray();
        while (reader.Read()) {
            reader.GetValues(buffer);
            for (int i = 0; i < len; ++i) {
                var value = buffer[i];
                if (value is DateTime dt) {
                    value = dt.Ticks;
                }
                rec[names[i]] = value == DBNull.Value ? null : value;
            }
            sw.Write(rec);
        }
    }
    
    
    0 回复  |  直到 2 年前
        1
  •  1
  •   Gokhan Atil    2 年前

    我把你的avro文件上传到了一个阶段,我可以在没有任何错误的情况下加载数据。

    create table testavro (id number, type varchar);
        
    copy into testavro from @mystage/LinkTypes.avro file_format = (type = avro) match_by_column_name = case_insensitive;
        
    select * from testavro;
    
    +----+-----------+
    | ID |   TYPE    |
    +----+-----------+
    |  1 | Linked    |
    |  3 | Duplicate |
    +----+-----------+
    
    ls @mystage;
    
    +---------------------------+------+----------------------------------+-------------------------------+
    |           name            | size |               md5                |         last_modified         |
    +---------------------------+------+----------------------------------+-------------------------------+
    | mystage/LinkTypes.avro.gz |  208 | 14747c0a8e9ab98cbd942feef5ef2702 | Wed, 27 Jul 2022 15:03:48 GMT |
    +---------------------------+------+----------------------------------+-------------------------------+
    

    所以我认为你应该重新上传你的文件。