0
0

FlinkSql使用ES sink并指定主键，为什么数据还是会被覆盖？

FlinkSql使用ES sink并指定主键，为什么数据还是会被覆盖？

1. 问题描述

根据ES connector文档中的描述，创建ES表并指定主键后将采用upsert模式。
但是在实际的使用过程中却发现部分数据仍然存在被直接覆盖的问题。
举个例子，假如ES中本来的数据如下图中示例1所示，在FlinkSql中要写入的数据如示例2所示，按照upsert的理解执行结果应该如示例3所示，但实际结果为示例4所示，即对整个json文档进行了覆盖，而不是部分字段的更新。
在这里插入图片描述
实际发生问题的场景如下：

  • 情形一：在任务启动前es中已经存在数据，任务启动后，发现部分主键key对应的json文档直接被覆盖成任务中新写入的数据
  • 情形二：任务中存在A和B两个insert into语句往同一个es表中写入数据，A和B除主键key之外其他写入的字段不完全相同，任务启动后，发现部分主键key对应的数据只包含A或B的结果。 上述两种情景的sql中均包含join、groupBy等操作。

2. 问题排查

针对上述问题，在网上查找一番后并未直接找到结果。大多数给出的答案正如官方文档所述，会执行upsert操作。
针对自己在实际使用过程中遇到的问题，尝试从源码中一探究竟。
接下来，首先需要找到源码的切入点，从切入点设置断点进行逐步分析。

2.1. 自定义sink原理

源码基于flink 1.14.4

根据官方文档中对自定义sink的描述，connector sink的的工作原理如下
在这里插入图片描述
元数据的内容由create table语句所定义，通过

CatalogTable

的实例表示，该实例表示

Catalog

中未解析的元数据，包含可以在create table语句中表示的所有特征，框架将其解析为

ResolvedCatalogTable

实例（一个经过验证的CatalogTable），然后再将其传递给

DynamicTableSinkFactory

，最终传入到

DynamicTableSink

，用于insert into语句的写入。

**

DynamicTableSinkFactory

负责提供connector的具体实现逻辑。用于将

CatalogTable

的元数据转化为

DynamicTableSink

。**
其接口中仅有一个方法，返回

DynamicTableSink

实例。

publicinterfaceDynamicTableSinkFactoryextendsDynamicTableFactory{/**
     * 根据CatalogTable的元数据转化为DynamicTableSink，CatalogTable通过org.apache.flink.table.factories.DynamicTableFactory.Context中获取
     * 实现应该此方法中执行create table中with选项的验证
     */DynamicTableSinkcreateDynamicTableSink(Context context);}

**

DynamicTableSink

负责将动态表sink到外部存储。**
动态表的内容可以被视为一个changelog，所有的更改都会连续写出，直到changelog结束为止。给定的

ChangelogMode

指示sink在运行时接受的一组更改。
该接口的实例可以看作是最终生成用于写入实际数据的具体运行时实现的工场。
最后planner将调用

getSinkRuntimeProvider

方法来获取运行时实现的提供程序，即

SinkRuntimeProvider

的实例。

**因此自定义sink的核心是通过

DynamicTableSink#getSinkRuntimeProvider

方法向planner提供一个

SinkRuntimeProvider

实例，该实例中包含了向外部存储写入数据的关键。**

publicinterfaceDynamicTableSink{/**
     * 返回sink在运行时接受的一组更改
     * planner可以提出建议，但sink拥有最终决定权。如果planner不支持此模式，它将引发错误。
     */ChangelogModegetChangelogMode(ChangelogMode requestedMode);/**
     * 返回用于写入数据的运行时实现，即SinkRuntimeProvider的实例
     * 表运行时独立于provider接口，要求sink实现接受内部数据结构（RowData）
     */SinkRuntimeProvidergetSinkRuntimeProvider(Context context);interfaceContext{booleanisBounded();<T>TypeInformation<T>createTypeInformation(DataType consumedDataType);DataStructureConvertercreateDataStructureConverter(DataType consumedDataType);}interfaceDataStructureConverterextendsRuntimeConverter{@NullableObjecttoExternal(@NullableObject internalStructure);}/**
     * 提供用于写入数据的实际运行时实现
     * SinkProvider 是sink的核心接口。
     * flink-table-api-java-bridge 中的 SinkFunctionProvider 和 OutputFormatProvider 可用于向后兼容。     *
     */interfaceSinkRuntimeProvider{// marker interface}}

了解了自定义sink的实现原理后，接下里看下es sink的具体实现。

2.2. ES sink实现

本文中使用的es connector版本为

<dependency><groupId>org.apache.flink</groupId><artifactId>flink-connector-elasticsearch6_2.11</artifactId><version>1.14.4</version></dependency>

DynamicTableSinkFactory

和

DynamicTableSink

在es sink中的实现分别为

Elasticsearch6DynamicSinkFactory

，

Elasticsearch6DynamicSink

。

Elasticsearch6DynamicSinkFactory

的实现如下，获取到scheme后对schema和配置参数进行验证，最后返回

Elasticsearch6DynamicSink

实例。

publicclassElasticsearch6DynamicSinkFactoryimplementsDynamicTableSinkFactory{@OverridepublicDynamicTableSinkcreateDynamicTableSink(Context context){// 从CatalogTable中获取schemaTableSchema tableSchema = context.getCatalogTable().getSchema();// 验证主键ElasticsearchValidationUtils.validatePrimaryKey(tableSchema);finalFactoryUtil.TableFactoryHelper helper =FactoryUtil.createTableFactoryHelper(this, context);// 得到formatfinalEncodingFormat<SerializationSchema<RowData>> format =
                helper.discoverEncodingFormat(SerializationFormatFactory.class,FORMAT_OPTION);

        helper.validate();Configuration configuration =newConfiguration();
        context.getCatalogTable().getOptions().forEach(configuration::setString);Elasticsearch6Configuration config =newElasticsearch6Configuration(configuration, context.getClassLoader());// 验证参数validate(config, configuration);// 最终返回Elasticsearch6DynamicSinkreturnnewElasticsearch6DynamicSink(
                format, config,TableSchemaUtils.getPhysicalSchema(tableSchema));}}

Elasticsearch6DynamicSink

的实现如下，

getChangelogMode

方法表示ES sink接受除UPDATE_BEFORE类型之外的其他更改。

getSinkRuntimeProvider

方法返回了

SinkFunctionProvider

实例，为

SinkRuntimeProvider

的子接口。
通过实现

SinkFunctionProvider

接口中的

SinkFunction<RowData> createSinkFunction()

方法，从而得到

SinkFunction

实例。

finalclassElasticsearch6DynamicSinkimplementsDynamicTableSink{@OverridepublicChangelogModegetChangelogMode(ChangelogMode requestedMode){ChangelogMode.Builder builder =ChangelogMode.newBuilder();for(RowKind kind : requestedMode.getContainedKinds()){// 不接受update_before类型的更改if(kind !=RowKind.UPDATE_BEFORE){
                builder.addContainedKind(kind);}}return builder.build();}@OverridepublicSinkFunctionProvidergetSinkRuntimeProvider(Context context){return()->{SerializationSchema<RowData> format =this.format.createRuntimeEncoder(context, schema.toRowDataType());// 负责生成真正和es交互的ActionRequest实例，如IndexRequest、UpdateRequest、DeleteRequest等finalRowElasticsearchSinkFunction upsertFunction =newRowElasticsearchSinkFunction(IndexGeneratorFactory.createIndexGenerator(config.getIndex(), schema),
                            config.getDocumentType(),
                            format,XContentType.JSON,REQUEST_FACTORY,KeyExtractor.createKeyExtractor(schema, config.getKeyDelimiter()));finalElasticsearchSink.Builder<RowData> builder =
                    builderProvider.createBuilder(config.getHosts(), upsertFunction);

            builder.setFailureHandler(config.getFailureHandler());
            builder.setBulkFlushMaxActions(config.getBulkFlushMaxActions());
            builder.setBulkFlushMaxSizeMb((int)(config.getBulkFlushMaxByteSize()>>20));
            builder.setBulkFlushInterval(config.getBulkFlushInterval());
            builder.setBulkFlushBackoff(config.isBulkFlushBackoffEnabled());
            config.getBulkFlushBackoffType().ifPresent(builder::setBulkFlushBackoffType);
            config.getBulkFlushBackoffRetries().ifPresent(builder::setBulkFlushBackoffRetries);
            config.getBulkFlushBackoffDelay().ifPresent(builder::setBulkFlushBackoffDelay);// we must overwrite the default factory which is defined with a lambda because of a bug// in shading lambda serialization shading see FLINK-18006if(config.getUsername().isPresent()&& config.getPassword().isPresent()&&!StringUtils.isNullOrWhitespaceOnly(config.getUsername().get())&&!StringUtils.isNullOrWhitespaceOnly(config.getPassword().get())){
                builder.setRestClientFactory(newAuthRestClientFactory(
                                config.getPathPrefix().orElse(null),
                                config.getUsername().get(),
                                config.getPassword().get()));}else{
                builder.setRestClientFactory(newDefaultRestClientFactory(config.getPathPrefix().orElse(null)));}// 最终是得到ElasticsearchSinkfinalElasticsearchSink<RowData> sink = builder.build();if(config.isDisableFlushOnCheckpoint()){
                sink.disableFlushOnCheckpoint();}return sink;};}}publicinterfaceSinkFunctionProviderextendsDynamicTableSink.SinkRuntimeProvider,ParallelismProvider{/** 返回SinkFunction的实例 */SinkFunction<RowData>createSinkFunction();}

因此planner在ES sink的实现中最终得到了

ElasticsearchSink

的实例。该实例负责调用ES相应API与ES进行数据交互。

2.3. 实际与ES的交互

通过

ElasticsearchSink

的继承关系，可以发现该类直接继承

ElasticsearchSinkBase

，并最终继承了

SinkFunction

。

在这里插入图片描述

SinkFunction

是实现自定义sink的根接口。每条写入sink的数据都将会调用该接口中的

invoke

方法。

ElasticsearchSinkBase

对此方法的实现如下，在其中通过调用

ElasticsearchSinkFunction#process

方法进行数据写入。

privatefinalElasticsearchSinkFunction<T> elasticsearchSinkFunction;@Overridepublicvoidinvoke(T value,Context context)throwsException{checkAsyncErrorsAndRequests();// 调用ElasticsearchSinkFunction的process方法
    elasticsearchSinkFunction.process(value,getRuntimeContext(), requestIndexer);}

在ES sink中

ElasticsearchSinkFunction

的唯一实现是

RowElasticsearchSinkFunction

类，其关键代码如下。

classRowElasticsearchSinkFunctionimplementsElasticsearchSinkFunction<RowData>{privatefinalRequestFactory requestFactory;@Override/**
     * 每条数据和ES实际交互的方法
     */publicvoidprocess(RowData element,RuntimeContext ctx,RequestIndexer indexer){// 根据每条数据的RowKind类型得到es api的UpdateRequest或DeleteRequest实例switch(element.getRowKind()){caseINSERT:caseUPDATE_AFTER:// insert和update_after时，进行更新processUpsert(element, indexer);break;caseUPDATE_BEFORE:caseDELETE:// update_before时，删除数据processDelete(element, indexer);break;default:thrownewTableException("Unsupported message kind: "+ element.getRowKind());}}// 对数据进行IndexRequest或UpdateRequest请求privatevoidprocessUpsert(RowData row,RequestIndexer indexer){finalbyte[] document = serializationSchema.serialize(row);finalString key = createKey.apply(row);if(key !=null){// 指定主键时，得到UpdateRequestfinalUpdateRequest updateRequest =
                    requestFactory.createUpdateRequest(
                            indexGenerator.generate(row), docType, key, contentType, document);
            indexer.add(updateRequest);}else{// 未指定主键时，得到IndexRequestfinalIndexRequest indexRequest =
                    requestFactory.createIndexRequest(
                            indexGenerator.generate(row), docType, key, contentType, document);
            indexer.add(indexRequest);}}// 对数据进行DeleteRequest请求privatevoidprocessDelete(RowData row,RequestIndexer indexer){finalString key = createKey.apply(row);finalDeleteRequest deleteRequest =
                requestFactory.createDeleteRequest(indexGenerator.generate(row), docType, key);
        indexer.add(deleteRequest);}}

其中

requestFactory.createUpdateRequest

、

requestFactory.createIndexRequest

、

requestFactory.createDeleteRequest

的方法分别如下

privatestaticclassElasticsearch6RequestFactoryimplementsRequestFactory{@OverridepublicUpdateRequestcreateUpdateRequest(String index,String docType,String key,XContentType contentType,byte[] document){returnnewUpdateRequest(index, docType, key).doc(document, contentType).upsert(document, contentType);}@OverridepublicIndexRequestcreateIndexRequest(String index,String docType,String key,XContentType contentType,byte[] document){returnnewIndexRequest(index, docType, key).source(document, contentType);}@OverridepublicDeleteRequestcreateDeleteRequest(String index,String docType,String key){returnnewDeleteRequest(index, docType, key);}}

到此，我们已经找到了ES sink和ES进行交互的具体逻辑，对于changelog中每条数据根据其

RowKind

类型决定执行INSERT、UPDATE还是DELETE请求。并且根据是否指定主键，细分为INSERT或UPDATE。

  • 指定主键，只会进行UPDATE和DELETE操作。
  • 未指定主键，只进行INSERT操作。

综上所述，在指定主键的情况下，如果changelog中只有INSERT和UPDATE_AFTER类型的数据时，那么仅调用UpdateRequest API将数据写入ES，数据只会执行Upsert操作。但是针对实际结果中发生数据覆盖的情况，只可能是存在DELETE类型的数据，导致其执行了processDelete方法，先将数据删除，随着changelog中数据不断到来再次执行processUpsert方法。最终导致结果中只保留了最新写入的数据，即看到的数据覆盖的情况。

DELETE类型的数据是否会真正生成？什么时机生成？

3. 验证演示

验证使用的示例代码如下，kafka中为随机生成的账单id从1到5，时间递增的账单数据。SQL加工逻辑为汇总每个id、每小时的账单金额，将结果写入到使用账单id作为主键的es表中。

publicclassEsConnectorSQLTest{publicstaticvoidmain(String[] args)throwsException{StreamExecutionEnvironment env =EnvConf.getEnvWithUI();StreamTableEnvironment tEnv =StreamTableEnvironment.create(env);
        env.setParallelism(1);// 创建source表，交易表
        tEnv.executeSql("CREATE TABLE transactions (\n"+"    account_id  BIGINT,\n"+"    amount      BIGINT,\n"+"    transaction_time TIMESTAMP(3),\n"+"    WATERMARK FOR transaction_time AS transaction_time - INTERVAL '5' SECOND\n"+") WITH (\n"+"    'connector' = 'kafka',\n"+"    'topic'     = 'transactions',\n"+"    'properties.bootstrap.servers' = '******',\n"+"    'scan.startup.mode' = 'latest-offset',\n"+"    'format'    = 'csv'\n"+")");// 创建sink表，消费报告表
        tEnv.executeSql("CREATE TABLE spend_report (\n"+"    account_id BIGINT,\n"+"    log_ts     TIMESTAMP(3),\n"+"    amount     BIGINT\n,"+"    PRIMARY KEY (account_id) NOT ENFORCED"+") WITH (\n"+"  'connector' = 'elasticsearch-6',\n"+"  'hosts' = ******n"+"  'index' = '******',\n"+"  'document-type' = '******',\n"+"  'sink.bulk-flush.max-actions' = '1',\n"+"  'username' = '******',\n"+"  'password' = '******',\n"+"  'format'   = 'json'\n"+")");// 从交易表中写数据到消费报告表中Table report =report(tEnv.from("transactions"));
        report.executeInsert("spend_report");// 通过如下代码观察select的执行结果数据，changelog// report.execute().print();}publicstaticTablereport(Table transactions){/*
         * 将每个账号在小时内的金额进行汇总
         * */return transactions.select(
                        $("account_id"),
                        $("transaction_time").floor(TimeIntervalUnit.HOUR).as("log_ts"),
                        $("amount")).groupBy($("account_id"), $("log_ts")).select(
                        $("account_id"),
                        $("log_ts"),
                        $("amount").sum().as("amount"));}}

通过在

RowElasticsearchSinkFunction#process

方法中设置断点，可以观察到对于同一个key的确产生了

-D

类型的数据，导致删除了ES中数据，后续再执行upsert方法时，由于之前的数据已经被删除，所以更新后只剩下了新写入的数据。

通过观察示例代码main方法中

report.execute().print()

结果，对于各个id首次聚合结果类型为

+I

，当数据执行更新时，先生成

-U

类型结果，将原先的结果回撤，然后紧跟着生成

+U

类型的最新结果。

selece的结果中并不会产生

-D

类型的数据，但是实际到达

RowElasticsearchSinkFunction#process

方法时却包含了

-D

类型的数据，因此

-D

类型的数据并不是有select语句直接得到的。

+----+----------------------+-------------------------+----------------------+
| op |           account_id |                  log_ts |               amount |
+----+----------------------+-------------------------+----------------------+
| +I |                    5 | 2565-04-12 05:00:00.000 |                  461 |
| +I |                    1 | 2565-04-12 05:00:00.000 |                  517 |
| +I |                    2 | 2565-04-12 05:00:00.000 |                  630 |
| +I |                    3 | 2565-04-12 05:00:00.000 |                  640 |
| +I |                    4 | 2565-04-12 05:00:00.000 |                  356 |
| -U |                    5 | 2565-04-12 05:00:00.000 |                  461 |
| +U |                    5 | 2565-04-12 05:00:00.000 |                 1140 |
| -U |                    1 | 2565-04-12 05:00:00.000 |                  517 |
| +U |                    1 | 2565-04-12 05:00:00.000 |                  688 |
| +I |                    2 | 2565-04-12 06:00:00.000 |                   35 |
| +I |                    3 | 2565-04-12 06:00:00.000 |                  524 |
| +I |                    4 | 2565-04-12 06:00:00.000 |                  290 |
| +I |                    5 | 2565-04-12 06:00:00.000 |                  612 |
| +I |                    1 | 2565-04-12 06:00:00.000 |                   12 |
| -U |                    2 | 2565-04-12 06:00:00.000 |                   35 |
| +U |                    2 | 2565-04-12 06:00:00.000 |                  903 |
| -U |                    3 | 2565-04-12 06:00:00.000 |                  524 |
| +U |                    3 | 2565-04-12 06:00:00.000 |                 1353 |

3.1. SinkMateriallizer任务

回到示例代码中，数据的加工逻辑是对数据进行Group by后的聚合处理，查看其作业DAG，可以发现有3个任务节点，分别是KafkaSource->GroupAggregate->SinkMateriallizer，前两个任务节点的作用容易理解，第三个名为SinkMaterializer节点除了最后将数据写入到ES外，还做了什么操作？算子名和算子中的

upsertMaterialize=[true]

表示什么含义？
在这里插入图片描述
根据算子名称并结合断点分析，数据会先进入到

GroupAggFunction#processElement

方法中执行聚合计算，并将聚合结果继续发送给了下游

SinkUpsertMaterializer#processElement

方法中，聚合结果最终在此方法中生成供上述

RowElasticsearchSinkFunction#process

方法中写入到ES的数据，接下来重点了解下

SinkUpsertMaterializer#processElement

的具体实现。

!!! note “”
SinkUpsertMaterializer类的作用是为了解决changelog中乱序流造成的结果不正确问题。

该维护了一个

List<RowData>

的状态，该状态中保存着changelog中相同key的结果数据，并且根据已保存到状态中的数据进行乱序处理。

publicclassSinkUpsertMaterializerextendsTableStreamOperator<RowData>implementsOneInputStreamOperator<RowData,RowData>{// Buffer of emitted insertions on which deletions will be applied first.// RowKind可能是+I或+U，并且在应用删除时将被忽略privatetransientValueState<List<RowData>> state;@OverridepublicvoidprocessElement(StreamRecord<RowData> element)throwsException{finalRowData row = element.getValue();List<RowData> values = state.value();if(values ==null){
            values =newArrayList<>(2);}switch(row.getRowKind()){caseINSERT:caseUPDATE_AFTER:
                row.setRowKind(values.isEmpty()?INSERT:UPDATE_AFTER);
                values.add(row);
                collector.collect(row);break;caseUPDATE_BEFORE:caseDELETE:finalint lastIndex = values.size()-1;finalint index =removeFirst(values, row);if(index ==-1){LOG.info(STATE_CLEARED_WARN_MSG);return;}if(values.isEmpty()){// 此处会将元素的RowKind重新赋值为-D
                    row.setRowKind(DELETE);
                    collector.collect(row);}elseif(index == lastIndex){// Last row has been removed, update to the second last onefinalRowData latestRow = values.get(values.size()-1);
                    latestRow.setRowKind(UPDATE_AFTER);
                    collector.collect(latestRow);}break;}if(values.isEmpty()){
            state.clear();}else{
            state.update(values);}}privateintremoveFirst(List<RowData> values,RowData remove){finalIterator<RowData> iterator = values.iterator();int i =0;while(iterator.hasNext()){finalRowData row = iterator.next();// 忽略RowKind类型，对RowData中其他数据进行比较
            remove.setRowKind(row.getRowKind());if(equaliser.equals(row, remove)){
                iterator.remove();return i;}
            i++;}return-1;}}

processElement方法大致处理逻辑如下流程图所示。
在这里插入图片描述
举个例子，

情形一：假如source表中有数据如左侧所示，经过示例sql处理后，输出了右侧的changelog结果
在这里插入图片描述

changelog
+----+------------+---------------+--------+
| op | account_id |      log_hour | amount |
+----+------------+---------------+--------+
| +I |          3 | 2565-04-12 05 |    640 |
| +I |          3 | 2565-04-12 06 |    524 |
| -U |          3 | 2565-04-12 06 |    524 |
| +U |          3 | 2565-04-12 06 |   1353 |

此种情况下changelog数据经过processElement方法后输出结果为：

  1. 当第一条数据到达后，数据中RowKind=+I，状态为空，输出RowKind=+I的结果到下游；
  2. 当第二条数据到达后，数据中RowKind=+I，状态不为空，输出RowKind=+U的结果到下游，这时状态中保存了changlog中第一条和第二条数据内容；
  3. 当第三条数据到达后，数据中RowKind=-U，状态集合中最大index=1（lastindex），依次比较集合中元素和当前数据内容是否相同，发现集合中第二条数据和当前数据一致，因此返回index=1，并将从集合中移除index=1的元素。此时集合中仅剩amount=640的一个元素； 接下来满足lastindex=index的条件，因此获取集合中最后一条数据，即amount=640的那条数据，输出Rowkind= +U的结果到下游；
  4. 当第四条数据到达后，数据中RowKind=+U，状态不为空，输出RowKind=+U的结果到下游。

情形二：假如source表中有数据如左侧所示，经过示例sql处理后，输出了右侧的changelog结果

在这里插入图片描述

changelog
+----+------------+---------------+--------+
| op | account_id |      log_hour | amount |
+----+------------+---------------+--------+
| +I |          3 | 2565-04-12 06 |    524 |
| -U |          3 | 2565-04-12 06 |    524 |
| +U |          3 | 2565-04-12 06 |   1353 |

此种情况下changelog数据经过processElement方法后输出结果为：

  1. 当第一条数据到达后，数据中RowKind=+I，状态为空，输出RowKind=+I的结果到下游；
  2. 当第二条数据到达后，数据中RowKind=-U，集合中最大index=0（lastindex），依次比较集合中元素和当前数据内容是否相同，发现集合中第一条数据和当前数据一致，因此返回index=0，并将从集合中移除index=0的元素。此时集合为空。 接下来满足了集合为空的条件，输出Rowkind=-D的结果到下游；
  3. 当第三条数据到达后，数据中RowKind=+U，状态为空，输出RowKind=+I的结果到下游。

4. 总结

导致写入到ES的数据从结果上看没有执行upsert而是执行了insert的根本原因是中间过程中执行了delete操作，将es中原始数据删除，然后继续写入了新数据。如果在delete执行后未继续写入新数据前停止任务，观察到的结果过将会是ES中数据条数减少。
而导致从ES中删除数据的原因是因为经过SinkUpsertMaterializer类处理后，将部分

RowKind=-U

的数据转化成了

RowKind=-D

类型的数据，从而触发ES sink执行

DeleteRequest

请求。

标签： 大数据 flink
本文转载自: https://blog.csdn.net/qq_33446500/article/details/138475389
版权归原作者 有数的编程笔记 所有， 如有侵权，请联系我们删除。
登录后发布评论

“FlinkSql使用ES sink并指定主键，为什么数据还是会被覆盖？”的评论:

还没有评论
关于作者
...
overfit同步小助手
文章同步
相关阅读

安全见闻（9）二进制安全

C++20 STL CookBook2 更强大的编译时 + 安全比较 + spaceship比较符

Linux：login shell和non-login shell以及其配置文件

使用git命令实现对gitee仓库的下载、更新、上传、上传更新操作。

network-magic：网络管理员必备的开源网络管理工具

git-- github的使用

结合Gitee、PicGo和Typora构建一个免费的个人笔记工具

文章导航