飞书API（10）：通过阿里云MaxCompute数仓入库 - 转为阿里云 DataFrame 再入库

news2025/2/12 22:01:37

一、引入

上一小节介绍了怎么入库到阿里云的 MaxCompute 数仓，其中涉及到 2 种入库方式，一种是转为阿里云的 DataFrame，然后类似 pandas 的 DataFrame 直接写入 MySQL 的方法，将数据写入表中；另外一种是转为列表，再写入 MaxCompute 表。上一小节主要对后者进行展开描述。
有粉丝私聊我说介绍下第一种，本文就来重点探讨下第一种的处理方式。
前面的数据我们通过 pandas 处理，所以这里涉及到将 pandas 的 DataFrame 转为 PyODPS 的 DataFrame 。
用Pandas DataFrame初始化时，PyODPS DataFrame 会尝试对 NUMPY OBJECT 或 STRING 类型进行推断。如果一整列都为空，则会报错。为避免报错，我们可以设置 unknown_as_string 值为 True，将这些列指定为 STRING 类型。如果 Pandas DataFrame 中包含 LIST 或 DICT 列，系统不会推断该列的类型，必须手动使用 as_type 指定类型。as_type参数类型必须是 DICT。
下面具体来探讨一下。

二、迭代入库逻辑

2.1 新增字段类型映射

阿里云的 DataFrame 拥有自己的类型系统，进行表初始化时，MaxCompute的类型会被转换成对应的DataFrame类型，以便支持更多类型的计算后端。
目前，DataFrame的执行后端支持MaxCompute SQL、Pandas和数据库（MySQL和Postgres）。数据类型的映射关系如下：

MaxCompute类型DataFrame类型
BIGINTINT64
DOUBLEFLOAT64
STRINGSTRING
DATETIMEDATETIME
BOOLEANBOOLEAN
DECIMALDECIMAL
ARRAY<VALUE_TYPE>LIST<VALUE_TYPE>
MAP<KEY_TYPE, VALUE_TYPE>DICT<KEY_TYPE, VALUE_TYPE>
当options.sql.use_odps2_extension=True时，还支持以下数据类型。
TINYINTINT8
SMALLINTINT16
INTINT32
FLOATFLOAT32

为了保证通用性和准确性，**建议通过 as_type 参数，指定数据类型。**避免由于 PyODPS DataFrame 推断有出入导致发生报错。
所以，需要加一层字段的映射关系。前面已经加了一层字段类型的映射关系，此处直接加上 PyODPS DataFrame 的类型即可：将double映射为float64，将array<string>映射为list<string>。
参考如下：

# 关联入库数据类型
data_type_map = [{"feishu_type": 1   ,"mc_type": "string"        ,"mc_df_type": "string"        }
                ,{"feishu_type": 2   ,"mc_type": "double"        ,"mc_df_type": "float64"       }
                ,{"feishu_type": 3   ,"mc_type": "string"        ,"mc_df_type": "string"        }
                ,{"feishu_type": 4   ,"mc_type": "array<string>" ,"mc_df_type": "list<string>"  }
                ,{"feishu_type": 5   ,"mc_type": "datetime"      ,"mc_df_type": "datetime"      }
                ,{"feishu_type": 7   ,"mc_type": "boolean"       ,"mc_df_type": "boolean"       }
                ,{"feishu_type": 11  ,"mc_type": "string"        ,"mc_df_type": "string"        }
                ,{"feishu_type": 13  ,"mc_type": "string"        ,"mc_df_type": "string"        }
                ,{"feishu_type": 15  ,"mc_type": "string"        ,"mc_df_type": "string"        }
                ,{"feishu_type": 17  ,"mc_type": "array<string>" ,"mc_df_type": "list<string>"  }
                ,{"feishu_type": 18  ,"mc_type": "array<string>" ,"mc_df_type": "list<string>"  }
                ,{"feishu_type": 19  ,"mc_type": "string"        ,"mc_df_type": "string"        }
                ,{"feishu_type": 20  ,"mc_type": "string"        ,"mc_df_type": "string"        }
                ,{"feishu_type": 21  ,"mc_type": "array<string>" ,"mc_df_type": "list<string>"  }
                ,{"feishu_type": 22  ,"mc_type": "string"        ,"mc_df_type": "string"        }
                ,{"feishu_type": 23  ,"mc_type": "string"        ,"mc_df_type": "string"        }
                ,{"feishu_type": 1001,"mc_type": "datetime"      ,"mc_df_type": "datetime"      }
                ,{"feishu_type": 1002,"mc_type": "datetime"      ,"mc_df_type": "datetime"      }
                ,{"feishu_type": 1003,"mc_type": "string"        ,"mc_df_type": "string"        }
                ,{"feishu_type": 1004,"mc_type": "string"        ,"mc_df_type": "string"        }
                ,{"feishu_type": 1005,"mc_type": "string"        ,"mc_df_type": "string"        }]

2.2 构建 PyODPS 的 DataFrame 的 astype

astype参数的值，可以传递一个字典，格式如：{“字段1”:“字段1类型”, “字段2”:“字段2类型”}。
在生成astype之前，通过merge_list()将字段映射fields_map、飞书列信息feishu_fields和数据类型映射data_type_map三者进行合并，返回store_fields_info_df，所以在生成astype时，只需要从store_fields_info_df中提取tb_field_name和mc_df_type，然后再加上record_id和last_modified_time即可。

def generate_astype(store_fields_info_df):
    """
    功能：对整合了飞书列入库的列、飞书列类型、和阿里云 DataFrame 数据类型的 pandas DataFrame，提取字段名和阿里云DataFrame 数据类型
    """
    #生成阿里云DataFrame的as_type，用于将处理好的数据转为阿里云DataFrame，格式：{'field_no': 'string', 'field_select': 'float64', 'field_text': 'datatime'}
    mc_df_astype = store_fields_info_df.set_index('tb_field_name').to_dict()['mc_df_type']
    #加上record_id
    mc_df_astype['record_id'] = 'string'
    #加一个表更新时间
    mc_df_astype['last_modified_time'] = 'DATETIME'
    # print(mc_df_astype)
    print('成功生成阿里云 DataFrame 的 as_type。方法：generate_astype')
    return mc_df_astype

2.3 生成 PyODPS DataFrame 并写入数仓表

def insert_mc_table(feishu_df, mc_table_name, mc_df_astype):
    """
    生成 PyODPS DataFrame 并将数据插入 Maxcompute 表
    """
    aliyun_df = DataFrame(feishu_df,as_type=mc_df_astype)
    aliyun_df.persist(mc_table_name)    
    print(f'成功将飞书数据写入 MaxCompute 数据表：{mc_table_name}。关联方法：insert_mc_table。')

2.4 迭代定制化函数

定制化函数也需要修改，原有的逻辑不用变，新增对astype的改动即可。
PyODPS DataFrame 包括以下类型：int8，int16，int32，int64，float32，float64，boolean，string，decimal，datetime，list，dict，不含 date，所以日期列不需要处理。

def custom_field(df_return, columns, columns_index, mc_df_astype):
    # 2.1 场景一：把数字入库为 int 类型
    # 修改 SQL 即可
    # cre_ddl = cre_ddl.replace('field_number float','field_number int')
    column = Column(name='field_number', type='bigint', comment=columns_index['field_number'][1])
    columns[columns_index['field_number'][0]] = column
    #修改mc_df_astype
    mc_df_astype['field_number'] = 'int64'
    
    # 2.2 场景二：把日期入库为 date 类型
    # 修改 df，MySQL会自动截断，Maxcompute不行，需要使用 x.date() 处理
    df_return['field_createdtime'] = df_return['field_createdtime'].apply(lambda x:x.date())
    # 修改 SQL
    # cre_ddl = cre_ddl.replace('field_createdtime datetime','field_createdtime date')
    column = Column(name='field_createdtime', type='date', comment=columns_index['field_createdtime'][1])
    columns[columns_index['field_createdtime'][0]] = column
    # #修改mc_df_astype
    # mc_df_astype['field_createdtime'] = 'date'
    
    # 2.3 场景三：日期给定默认最大值
    # 修改 df 即可
    #默认值改为 2222-01-01 00:00:00
    mask = df_return['field_date'] == pd.Timestamp('1970-01-01 08:00:01')
    df_return.loc[mask, 'field_date'] = pd.Timestamp('2222-01-01 00:00:00')
    
    # 2.4 场景四：公式保留具体值
    # 修改 df
    # 修改 SQL
    df_return['field_numformula'] = df_return['field_numformula'].apply(lambda x:json.loads(x)['value'][0])
    # cre_ddl = cre_ddl.replace('field_numformula varchar(256)','field_numformula int')
    column = Column(name='field_numformula', type='bigint', comment=columns_index['field_numformula'][1])
    columns[columns_index['field_numformula'][0]] = column
    #修改mc_df_astype
    mc_df_astype['field_numformula'] = 'int64'

    # 创建新的 schema
    schema = Schema(columns=columns)
    print('定制函数打印数据和建表语句')
    print('----------------------------------------------\n', df_return[['field_number','field_createdtime','field_date','field_numformula']].head(5))
    print('----------------------------------------------\n', schema.columns)
    return df_return, schema

三、整合代码

将上一小节最终的整合代码结合上面的三项内容进行修改。

  • 修改data_type_map：加上键值对mc_df_type；
  • 使用generate_astype()+insert_mc_table() 替换replace_nan_with_none()和insert_mc_table()；
  • 替换定制化函数的内容

最终参考代码如下：


import requests
import json
import datetime
import pandas as pd
from sqlalchemy import create_engine, text
from urllib.parse import urlparse, parse_qs
from odps.models import Schema, Column
import math

def get_table_params(bitable_url):
    # bitable_url = "https://feishu.cn/base/aaaaaaaa?table=tblccc&view=vewddd"
    parsed_url = urlparse(bitable_url)              #解析url：(ParseResult(scheme='https', netloc='feishu.cn', path='/base/aaaaaaaa', params='', query='table=tblccc&view=vewddd', fragment='')
    query_params = parse_qs(parsed_url.query)       #解析url参数：{'table': ['tblccc'], 'view': ['vewddd']}
    app_token = parsed_url.path.split('/')[-1]
    table_id = query_params.get('table', [None])[0]
    view_id = query_params.get('view', [None])[0]
    print(f'成功解析链接，app_token：{app_token}，table_id：{table_id}，view_id：{view_id}。关联方法：get_table_params。')
    return app_token, table_id, view_id

def get_tenant_access_token(app_id, app_secret):
    url = "https://open.feishu.cn/open-apis/auth/v3/tenant_access_token/internal"
    payload = json.dumps({
        "app_id": app_id,
        "app_secret": app_secret
    })
    headers = {'Content-Type': 'application/json'}
    response = requests.request("POST", url, headers=headers, data=payload)
    tenant_access_token = response.json()['tenant_access_token']
    print(f'成功获取tenant_access_token：{tenant_access_token}。关联函数：get_table_params。')
    return tenant_access_token

def get_bitable_datas(tenant_access_token, app_token, table_id, view_id, page_token='', page_size=20):
    url = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{app_token}/tables/{table_id}/records/search?page_size={page_size}&page_token={page_token}&user_id_type=user_id"
    payload = json.dumps({"view_id": view_id})
    headers = {
        'Content-Type': 'application/json',
        'Authorization': f'Bearer {tenant_access_token}'
    }
    response = requests.request("POST", url, headers=headers, data=payload)
    print(f'成功获取page_token为【{page_token}】的数据。关联函数：get_bitable_datas。')
    return response.json()


def get_all_bitable_datas(tenant_access_token, app_token, table_id, view_id, page_token='', page_size=20):
    has_more = True
    feishu_datas = []
    while has_more:
        response = get_bitable_datas(tenant_access_token, app_token, table_id, view_id, page_token, page_size)
        if response['code'] == 0:
            page_token = response['data'].get('page_token')
            has_more = response['data'].get('has_more')
            # print(response['data'].get('items'))
            # print('\n--------------------------------------------------------------------\n')
            feishu_datas.extend(response['data'].get('items'))
        else:
            raise Exception(response['msg'])
    print(f'成功获取飞书多维表所有数据，返回 feishu_datas。关联函数：get_all_bitable_datas。')
    return feishu_datas

def get_bitable_fields(tenant_access_token, app_token, table_id, page_size=500):
    url = f"https://open.feishu.cn/open-apis/bitable/v1/apps/{app_token}/tables/{table_id}/fields?page_size={page_size}"
    payload = ''
    headers = {'Authorization': f'Bearer {tenant_access_token}'}
    
    response = requests.request("GET", url, headers=headers, data=payload)
    field_infos = response.json().get('data').get('items')
    print('成功获取飞书字段信息，关联函数：get_bitable_fields。')
    return field_infos

def merge_list(ls_from, ls_join, on=None, left_on=None, right_on=None):
    """将两个[{},{}]结构的数据合并"""
    df_from = pd.DataFrame(ls_from)
    df_join = pd.DataFrame(ls_join)
    if on is not None:
        df_merge = df_from.merge(df_join, how='left', on=on)
    else:
        df_merge = df_from.merge(df_join, how='left', left_on=left_on, right_on=right_on) # , suffixes=('', '_y')
    print(f'成功合并列表或DataFrame。关联方法：merge_list。')
    return df_merge


def extract_key_fields(feishu_datas, store_fields_info_df):
    """处理飞书数据类型编号的数据"""
    print('开始处理飞书多维表关键字段数据...')
    # 需要record_id 和 订单号，用于和数据库数据匹配
    df_feishu = pd.DataFrame(feishu_datas)
    df_return = pd.DataFrame()
    
    #根据列的数据类型，分别处理对应的数据。注意：仅返回以下列举的数据类型，如果fields_map的内容包含按钮、流程等数据类型的飞书列，忽略。
    for index, row in store_fields_info_df.iterrows():
        if row['type'] == 1:       #文本
            df_return[row['tb_field_name']] = df_feishu['fields'].apply(lambda x:x.get(row['field_name'],[{}])[0].get('text'))
        
        elif row['type'] in (2, 3, 4, 7, 13, 1005):  #数字、单选、多选、复选框、手机号、自动编号
            df_return[row['tb_field_name']] = df_feishu['fields'].apply(lambda x:x.get(row['field_name']))
        
        elif row['type'] in (5, 1001, 1002):         #日期(包含创建和更新)，需要加 8 小时，即 8*60*60*1000=28800 秒
            df_return[row['tb_field_name']] = pd.to_datetime(df_feishu['fields'].apply(lambda x:28800 + int(x.get(row['field_name'],1000)/1000)), unit='s')
        
        elif row['type'] in(11, 23, 1003, 1004):       #人员、群组、创建人、修改人，遍历取name
            df_return[row['tb_field_name']] = df_feishu['fields'].apply(lambda x: ','.join([i.get('name') for i in x.get(row['field_name'],[{"name":""}])]))  # 需要遍历

        elif row['type'] == 15:    #链接
            df_return[row['tb_field_name']] = df_feishu['fields'].apply(lambda x:x.get(row['field_name'],{}).get('link'))
        
        elif row['type'] == 17:    #附件，遍历取url
            df_return[row['tb_field_name']] = df_feishu['fields'].apply(lambda x:[i.get('url') for i in x.get(row['field_name'],[{}])]) #需要遍历
    
        elif row['type'] in(18, 21):    #单向关联、双向关联，取link_record_ids
            df_return[row['tb_field_name']] = df_feishu['fields'].apply(lambda x:x.get(row['field_name'],{}).get('link_record_ids'))
    
        elif row['type'] in(19, 20):    #查找引用和公式
            df_return[row['tb_field_name']] = df_feishu['fields'].apply(lambda x:json.dumps(x.get(row['field_name'])))
    
        elif row['type'] == 22:    #地理位置
            df_return[row['tb_field_name']] = df_feishu['fields'].apply(lambda x:x.get(row['field_name'],{}).get('location'))
            
    #加上record_id
    df_return['record_id'] = df_feishu.record_id
    #加上表更新字段
    df_return['last_modified_time'] = datetime.datetime.now()
    print(f'成功提取入库字段的数据。关联方法：extract_key_fields。')
    return df_return

def generate_create_schema(store_fields_info_df):
    columns = []
    columns_index = {}
    for index, row in store_fields_info_df.iterrows():
        name = row['tb_field_name']; type=row['mc_type']; comment=row['feishu_field_name']
        # print(name,type,comment)
        columns.append(Column(name=name, type=type, comment=comment))
        columns_index[name] = (index, comment)
    columns.append(Column(name='record_id', type='string', comment='飞书表行record_id'))
    columns.append(Column(name='last_modified_time', type='datetime', comment='数据更新时间'))
    schema = Schema(columns=columns)
    print(f'成功生成 MaxCompute 建表 schema。关联方法：generate_create_schema。')
    return schema, columns, columns_index

def cre_mc_table(db_table_name, schema):
    if o.exist_table(db_table_name):
        print(f'表单 {db_table_name} 已存在，不需要新建。关联方法：cre_mc_table。')
    else:
        table = o.create_table(db_table_name, schema, if_not_exists=True)
        print(f'成功创建 MaxCompute 表：{db_table_name}。关联方法：cre_mc_table。')

def generate_astype(store_fields_info_df):
    """
    功能：对整合了飞书列入库的列、飞书列类型、和阿里云 DataFrame 数据类型的 pandas DataFrame，提取字段名和阿里云DataFrame 数据类型
    """
    #生成阿里云DataFrame的as_type，用于将处理好的数据转为阿里云DataFrame，格式：{'field_no': 'string', 'field_select': 'float64', 'field_text': 'datatime'}
    mc_df_astype = store_fields_info_df.set_index('tb_field_name').to_dict()['mc_df_type']
    #加上record_id
    mc_df_astype['record_id'] = 'string'
    #加一个表更新时间
    mc_df_astype['last_modified_time'] = 'DATETIME'
    # print(mc_df_astype)
    print('成功生成阿里云 DataFrame 的 as_type。方法：generate_astype')
    return mc_df_astype

def insert_mc_table(feishu_df, mc_table_name, mc_df_astype):
    """
    生成 PyODPS DataFrame 并将数据插入 Maxcompute 表
    """
    aliyun_df = DataFrame(feishu_df,as_type=mc_df_astype)
    aliyun_df.persist(mc_table_name)    
    print(f'成功将飞书数据写入 MaxCompute 数据表：{mc_table_name}。关联方法：insert_mc_table。')

def custom_field(df_return, columns, columns_index, mc_df_astype):
    # 2.1 场景一：把数字入库为 int 类型
    # 修改 SQL 即可
    # cre_ddl = cre_ddl.replace('field_number float','field_number int')
    column = Column(name='field_number', type='bigint', comment=columns_index['field_number'][1])
    columns[columns_index['field_number'][0]] = column
    #修改mc_df_astype
    mc_df_astype['field_number'] = 'int64'
    
    # 2.2 场景二：把日期入库为 date 类型
    # 修改 df，MySQL会自动截断，Maxcompute不行，需要使用 x.date() 处理
    df_return['field_createdtime'] = df_return['field_createdtime'].apply(lambda x:x.date())
    # 修改 SQL
    # cre_ddl = cre_ddl.replace('field_createdtime datetime','field_createdtime date')
    column = Column(name='field_createdtime', type='date', comment=columns_index['field_createdtime'][1])
    columns[columns_index['field_createdtime'][0]] = column
    # #修改mc_df_astype
    # mc_df_astype['field_createdtime'] = 'date'
    
    # 2.3 场景三：日期给定默认最大值
    # 修改 df 即可
    #默认值改为 2222-01-01 00:00:00
    mask = df_return['field_date'] == pd.Timestamp('1970-01-01 08:00:01')
    df_return.loc[mask, 'field_date'] = pd.Timestamp('2222-01-01 00:00:00')
    
    # 2.4 场景四：公式保留具体值
    # 修改 df
    # 修改 SQL
    df_return['field_numformula'] = df_return['field_numformula'].apply(lambda x:json.loads(x)['value'][0])
    # cre_ddl = cre_ddl.replace('field_numformula varchar(256)','field_numformula int')
    column = Column(name='field_numformula', type='bigint', comment=columns_index['field_numformula'][1])
    columns[columns_index['field_numformula'][0]] = column
    #修改mc_df_astype
    mc_df_astype['field_numformula'] = 'int64'

    # 创建新的 schema
    schema = Schema(columns=columns)
    print('定制函数打印数据和建表语句')
    print('----------------------------------------------\n', df_return[['field_number','field_createdtime','field_date','field_numformula']].head(5))
    print('----------------------------------------------\n', schema.columns)
    return df_return, schema, mc_df_astype

def main(mc_table_name, bitable_url, fields_map):
    # 基本配置
    app_token, table_id, view_id = get_table_params(bitable_url)
    app_id = 'your_app_id'
    app_secret = 'your_app_secret'
    tenant_access_token = get_tenant_access_token(app_id, app_secret)
    page_size = 50
    
    # 获取飞书多维表所有数据
    feishu_datas = get_all_bitable_datas(tenant_access_token, app_token, table_id, view_id, page_size=page_size)

    #获取飞书字段信息
    feishu_fields = get_bitable_fields(tenant_access_token, app_token, table_id)

    # 以 fields_map 为准关联数据
    store_fields_info_df = merge_list(fields_map, feishu_fields, left_on='feishu_field_name', right_on='field_name')

    # 处理入库字段数据
    feishu_df = extract_key_fields(feishu_datas, store_fields_info_df)
    
    # 关联入库数据类型
    data_type_map = [{"feishu_type": 1   ,"mc_type": "string"        ,"mc_df_type": "string"        }
                    ,{"feishu_type": 2   ,"mc_type": "double"        ,"mc_df_type": "float64"       }
                    ,{"feishu_type": 3   ,"mc_type": "string"        ,"mc_df_type": "string"        }
                    ,{"feishu_type": 4   ,"mc_type": "array<string>" ,"mc_df_type": "list<string>"  }
                    ,{"feishu_type": 5   ,"mc_type": "datetime"      ,"mc_df_type": "datetime"      }
                    ,{"feishu_type": 7   ,"mc_type": "boolean"       ,"mc_df_type": "boolean"       }
                    ,{"feishu_type": 11  ,"mc_type": "string"        ,"mc_df_type": "string"        }
                    ,{"feishu_type": 13  ,"mc_type": "string"        ,"mc_df_type": "string"        }
                    ,{"feishu_type": 15  ,"mc_type": "string"        ,"mc_df_type": "string"        }
                    ,{"feishu_type": 17  ,"mc_type": "array<string>" ,"mc_df_type": "list<string>"  }
                    ,{"feishu_type": 18  ,"mc_type": "array<string>" ,"mc_df_type": "list<string>"  }
                    ,{"feishu_type": 19  ,"mc_type": "string"        ,"mc_df_type": "string"        }
                    ,{"feishu_type": 20  ,"mc_type": "string"        ,"mc_df_type": "string"        }
                    ,{"feishu_type": 21  ,"mc_type": "array<string>" ,"mc_df_type": "list<string>"  }
                    ,{"feishu_type": 22  ,"mc_type": "string"        ,"mc_df_type": "string"        }
                    ,{"feishu_type": 23  ,"mc_type": "string"        ,"mc_df_type": "string"        }
                    ,{"feishu_type": 1001,"mc_type": "datetime"      ,"mc_df_type": "datetime"      }
                    ,{"feishu_type": 1002,"mc_type": "datetime"      ,"mc_df_type": "datetime"      }
                    ,{"feishu_type": 1003,"mc_type": "string"        ,"mc_df_type": "string"        }
                    ,{"feishu_type": 1004,"mc_type": "string"        ,"mc_df_type": "string"        }
                    ,{"feishu_type": 1005,"mc_type": "string"        ,"mc_df_type": "string"        }]

    store_fields_info_df = merge_list(store_fields_info_df, data_type_map, left_on='type', right_on='feishu_type')

    # 生成 MaxCompute schema
    schema, columns, columns_index = generate_create_schema(store_fields_info_df)

    # 生成 DataFrame astype
    mc_df_astype = generate_astype(store_fields_info_df)
    
    # 定制化
    feishu_df, schema, mc_df_astype = custom_field(feishu_df, columns, columns_index, mc_df_astype)

    # 建 MaxCompute 数据表
    cre_mc_table(mc_table_name, schema)

    # MaxCompute 表插入数据
    insert_mc_table(feishu_df, mc_table_name, mc_df_astype)
    
    print(f'成功将飞书多维表（{bitable_url}）的数据入库到 mysql 数据表：{mc_table_name}。')

if __name__ == '__main__':
    mc_table_name = 'for_ods.feishu_data_type_test'
    bitable_url = "https://forchangesz.feishu.cn/base/SpY3b9LMFaodpOsE0kdcGEyonbg?table=tbl5BZE0Aubjz5Yy&view=vewDM4NGlP"
    fields_map = [{'tb_field_name': 'field_text','feishu_field_name': '文本'}
            ,{'tb_field_name': 'field_email','feishu_field_name': 'email'}
            ,{'tb_field_name': 'field_select','feishu_field_name': '单选'}
            ,{'tb_field_name': 'field_mobile','feishu_field_name': '电话号码'}
            ,{'tb_field_name': 'field_no','feishu_field_name': '自动编号'}
            ,{'tb_field_name': 'field_member1','feishu_field_name': '人员1'}
            ,{'tb_field_name': 'field_group1','feishu_field_name': '群组1'}
            ,{'tb_field_name': 'field_creator','feishu_field_name': '创建人'}
            ,{'tb_field_name': 'field_modifier','feishu_field_name': '修改人'}
            ,{'tb_field_name': 'field_member2','feishu_field_name': '人员2'}
            ,{'tb_field_name': 'field_group2','feishu_field_name': '群组2'}
            ,{'tb_field_name': 'field_url','feishu_field_name': '超链接'}
            ,{'tb_field_name': 'field_location','feishu_field_name': '地理位置'}
            ,{'tb_field_name': 'field_findnum','feishu_field_name': '查找引用数值'}
            ,{'tb_field_name': 'field_numformula','feishu_field_name': '数字公式'}
            ,{'tb_field_name': 'field_number','feishu_field_name': '数字'}
            ,{'tb_field_name': 'field_progress','feishu_field_name': '进度'}
            ,{'tb_field_name': 'field_money','feishu_field_name': '货币'}
            ,{'tb_field_name': 'field_Rating','feishu_field_name': '评分'}
            ,{'tb_field_name': 'field_bool','feishu_field_name': '复选框'}
            ,{'tb_field_name': 'field_date','feishu_field_name': '日期'}
            ,{'tb_field_name': 'field_createdtime','feishu_field_name': '创建时间'}
            ,{'tb_field_name': 'field_updatedtime','feishu_field_name': '更新时间'}
            ,{'tb_field_name': 'field_mulselect','feishu_field_name': '多选'}
            ,{'tb_field_name': 'field_singleunion','feishu_field_name': '单向关联'}
            ,{'tb_field_name': 'field_doubleunion','feishu_field_name': '双向关联'}
            ,{'tb_field_name': 'field_file','feishu_field_name': '附件'}
            ]
    main(mc_table_name, bitable_url, fields_map)

最终执行的结果和上一小节一致，参考如下：
image.png

image.png

image.png

四、小结

本文探讨了怎么通过 PyODPS 的 DataFrame 将飞书数据入库，主要涉及四点：新增 PyODPS 的 DataFrame 的数据类型映射、定义 astype、将飞书数据转为 PyODPS 的 DataFrame 并入库和定制化中新增 astype 的修改。

PyODPS 的 DataFrame 更多用于数据科学计算，方便将分析的结果数据保存到表中，此处仅用它作为一个中间桥梁，将 pandas 的 DataFrame 和数仓表连接起来。

本文来自互联网用户投稿，该文观点仅代表作者本人，不代表本站立场。本站仅提供信息存储空间服务，不拥有所有权，不承担相关法律责任。如若转载，请注明出处：http://www.coloradmin.cn/o/1701638.html

如若内容造成侵权/违法违规/事实不符，请联系多彩编程网进行投诉反馈，一经查实，立即删除！

相关文章

【openlayers系统学习】4.3VectorTile 功能交互（指针悬停在要素上时，绘制矩形框）

【openlayers系统学习】4.3VectorTile 功能交互（指针悬停在要素上时，绘制矩形框）

三、 VectorTile 功能交互&#xff08;指针悬停在要素上时&#xff0c;绘制矩形框&#xff09; 矢量切片的好处是我们可以与要素交互&#xff0c;因为我们在客户端上有数据。但需要注意的一件事是矢量切片针对渲染进行了优化。这意味着要素仅包含过滤和渲染所需的属性&#xf…
阅读更多...
如何知道huggingface/modelscope的大模型的模型层名字

如何知道huggingface/modelscope的大模型的模型层名字

下载模型后&#xff0c;有个文件叫model.safetensors.index.json&#xff0c;里面有。 你下载的大模型位置在用户名/.cache/huggingface/hub/大模型名差不多这个路径。 或者直接print(parameters.name)&#xff0c;但是这样打出来特别多&#xff0c;很难看。差不多这样写&am…
阅读更多...
Android 10.0 锁屏页面弹窗功能实现

Android 10.0 锁屏页面弹窗功能实现

1.前言 在10.0的系统rom产品定制化开发中，在定制化开发中，需要在锁屏页面弹窗功能，当收到某些信息的时候，需要添加 悬浮窗锁屏页面也同样需要弹窗功能，接下来就分析下相关功能，然后实现功能 2.锁屏页面弹窗功能实现的核心类 frameworks\base\core\java\android\view\Wi…
阅读更多...
【Chrono Engine学习总结】6-创建自定义场景-6.2-chrono中的光线设置

【Chrono Engine学习总结】6-创建自定义场景-6.2-chrono中的光线设置

由于Chrono的官方教程在一些细节方面解释的并不清楚&#xff0c;自己做了一些尝试&#xff0c;做学习总结。 上一篇文章中&#xff0c;自己【用sketchup重建了三维场景】&#xff0c;但导入chrono中颜色很不正确&#xff0c;几乎都是白色的&#xff0c;但也不是完全白色。经过…
阅读更多...
2024年上半年软考什么时候查成绩？附查询流程

2024年上半年软考什么时候查成绩？附查询流程

考试一旦结束&#xff0c;并不意味着与考试相关的事情也就结束了。2024年上半年信息系统项目管理师等软考考试结束后&#xff0c;我们还需要关注考后和证书相关的事情&#xff0c;比如成绩查询、证书领取等等。 2024年上半年软考成绩查询 查询时间&#xff1a;预计在2024年7月…
阅读更多...
优优嗨聚集团：快速摆脱个人债务束缚的秘诀

优优嗨聚集团：快速摆脱个人债务束缚的秘诀

在快节奏的现代生活中&#xff0c;个人债务问题时常困扰着许多人。面对日益增长的债务压力&#xff0c;如何快速有效地处理成为众多人的迫切需求。本文将为你提供一套实用的债务清零攻略&#xff0c;帮助你摆脱债务的束缚&#xff0c;重获财务自由。 一、认清债务现状&#xff…
阅读更多...
艾体宝洞察 | Redis Enterprise对比ElastiCache

艾体宝洞察 | Redis Enterprise对比ElastiCache

选择缓存数据库时&#xff0c;如何在Amazon ElastiCache和Redis Enterprise之间做出选择&#xff0c;应当考虑哪些标准&#xff1f; ElastiCache 通常可以满足基本的缓存需求&#xff0c;因此是一种适合初始阶段的解决方案。但随着使用量的增加&#xff0c;ElastiCache很快会变…
阅读更多...
Octo 精武门？ ：开源的通用机器人模型

Octo 精武门？ ：开源的通用机器人模型

大模型技术论文不断&#xff0c;每个月总会新增上千篇。本专栏精选论文重点解读&#xff0c;主题还是围绕着行业实践和工程量产。若在某个环节出现卡点&#xff0c;可以回到大模型必备腔调重新阅读。而最新科技&#xff08;Mamba,xLSTM,KAN&#xff09;则提供了大模型领域最新技…
阅读更多...
[RK3588-Android12] 关于ES8388 喇叭+PDM回采 4+2配置

[RK3588-Android12] 关于ES8388 喇叭+PDM回采 4+2配置

问题描述&#xff1a; ES8388 喇叭PDM回采 42配置如下&#xff1a; 解决方案&#xff1a; // MICpdmics: dummy-codec {status "okay";compatible "rockchip,dummy-codec";#sound-dai-cells <0>;};// MICpdm_mic_array: pdm-mic-array {status …
阅读更多...
【Linux002】cURL接口测试常用命令总结（已更新）

【Linux002】cURL接口测试常用命令总结（已更新）

1.熟悉、梳理、总结项目研发实战中的cURL接口测试常用命令&#xff0c;刚好是最近研发遇到的问题。 2.欢迎点赞、关注、批评、指正&#xff0c;互三走起来&#xff0c;小手动起来&#xff01; 文章目录 1.cURL简要介绍2. cURL常用命令清单3. 测试命令行4. 部分效果示例&#xf…
阅读更多...
教育新篇章：AI工具Sora引领学习新趋势

教育新篇章：AI工具Sora引领学习新趋势

Sora——这个让人在24年初引爆AI圈的新产品&#xff0c;它究竟会如何改变我们的教育领域呢&#xff1f; 从gpt到Sora&#xff0c;从对话型的ai到游戏和短剧制作的新风口&#xff0c;我们从23年到24年一个接一个地被震惊&#xff01; Sora能够根据文本提示生成高质量的视频内容…
阅读更多...
哈工大操作系统—多进程图像

哈工大操作系统—多进程图像

操作系统管理cpu的时候 引出多进程 多进程为操作系统的核心 把程序存放在内存里 设置一个pc地址 发出取指的命令 cpu工作 把指令通过总线传到
阅读更多...
前端工具vscode 提交代码git操作

前端工具vscode 提交代码git操作

一 常规操作 1. 相当于git add . 2.输入修改备注&#xff0c;点击提交 3.git push 推送&#xff08;完成这步&#xff0c;本地代码就成功推送到了远程git&#xff09; 二、切换分支 三 比较与之前版本的改动内容 四。合并分支&#xff08;分两步&#xff0c;两截图&#xff0…
阅读更多...
图像处理之DBSCAN算法（C++）

图像处理之DBSCAN算法（C++）

图像处理之DBSCAN算法&#xff08;C&#xff09; 文章目录 图像处理之DBSCAN算法&#xff08;C&#xff09;前言一、DBSCAN算法原理二、代码实现总结 前言 DBSCAN聚类算法是一种无监督的数据分类方法&#xff0c;该算法不需要训练数据就可以实现对数据的分类。 一、DBSCAN算法…
阅读更多...
最最最重要的集群参数配置（上）no.7

最最最重要的集群参数配置（上）no.7

我希望通过两期内容把这些重要的配置讲清楚。严格来说这些配置并不单单指Kafka服务器端的配置&#xff0c;其中既有Broker端参数&#xff0c;也有主题&#xff08;后面我用我们更熟悉的Topic表示&#xff09;级别的参数、JVM端参数和操作系统级别的参数。 需要你注意的是&…
阅读更多...
家政上门服务小程序，客商紧密连系的作用是什么

家政上门服务小程序，客商紧密连系的作用是什么

家政服务拓展速度很快&#xff0c;大小城市都有不少品牌门店&#xff0c;其涵盖项目多样化&#xff0c;使得部分年轻人和老年人长期消费需要&#xff0c;商家与客户都需要完善的路径进行长期合作。 运用【雨科】平台搭建家政上门服务预约小程序&#xff0c;客户随时预约服务、…
阅读更多...
技术前沿：三品PLM系统引领工程变更管理新趋势

技术前沿：三品PLM系统引领工程变更管理新趋势

引言 在当今快速变化的制造行业&#xff0c;产品生命周期管理&#xff08;PLM&#xff09;系统已成为企业不可或缺的工具之一。PLM系统不仅帮助企业优化产品开发流程&#xff0c;还对工程变更管理&#xff08;ECM&#xff09;起着至关重要的作用。本文将探讨PLM系统在工程变更…
阅读更多...
高职高校实训教学实验室管理系统一体化

高职高校实训教学实验室管理系统一体化

盛元广通高职高校实训教学实验室管理系统一体化是确保实验教学有序进行的关键环节。通过更加科学 、有效、合理的管理&#xff0c;明确排课原则、收集课程信息、评估实验室资源、制定排课计划、冲突检测与调整、发布排课信息、调课管理以及数据统计与分析等措施。实现了实验室资…
阅读更多...
Go语言

Go语言

文章目录 Go语言特点应用领域Go语言和Java语言的对比JavaGo 总结 Go语言安装配置环境变量Hello World Go语言 Go语言全称Golanguage&#xff0c;Go&#xff08;又称 Golang&#xff09;是 Google 的 Robert Griesemer&#xff0c;Rob Pike 及 Ken Thompson 开发的一种静态强类…
阅读更多...
数组Array

数组Array

数组的基本用法 概念 数组是有序元素序列。如果将若干个数据类型相同的变量的集合命名&#xff0c;那么该命名就是数组名。数组元素的重点是类型相同并且连续在内存中存放的数据。 定义格式 存储类型 数据类型 数组名 [ 元素个数 ] &#xff1b; 例如&#xff1a; &#x…
阅读更多...
推荐文章
最新文章

Copyright @ 2022~2023