diff --git a/metadata-ingestion/README.md b/metadata-ingestion/README.md
index 989031dcd8..a619f5f32c 100644
--- a/metadata-ingestion/README.md
+++ b/metadata-ingestion/README.md
@@ -173,6 +173,23 @@ source:
     # table_pattern is same as above
 ```
 
+## Snowflake `snowflake`
+Extracts:
+- List of databases, schema, and tables
+- Column types associated with each table
+
+Extra requirements: `pip install snowflake-sqlalchemy`
+
+```yml
+source:
+  type: snowflake
+  config:
+    username: user
+    password: pass
+    host_port: account_name
+    # table_pattern is same as above
+```
+
 ## File `file`
 Pulls metadata from a previously generated file. Note that the file sink
 can produce such files, and a number of samples are included in the
diff --git a/metadata-ingestion/src/gometa/ingestion/source/__init__.py b/metadata-ingestion/src/gometa/ingestion/source/__init__.py
index 22e5e24d53..d49edb5b3f 100644
--- a/metadata-ingestion/src/gometa/ingestion/source/__init__.py
+++ b/metadata-ingestion/src/gometa/ingestion/source/__init__.py
@@ -10,12 +10,14 @@ from .mssql import SQLServerSource
 from .mysql import MySQLSource
 from .hive import HiveSource
 from .postgres import PostgresSource
+from .snowflake import SnowflakeSource
 
 source_class_mapping: Dict[str, Type[Source]] = {
     "mssql": SQLServerSource,
     "mysql": MySQLSource,
     "hive": HiveSource,
     "postgres": PostgresSource,
+    "snowflake": SnowflakeSource,
     "kafka": KafkaSource,
     # "ldap": LDAPSource,
     "file": MetadataFileSource,
diff --git a/metadata-ingestion/src/gometa/ingestion/source/snowflake.py b/metadata-ingestion/src/gometa/ingestion/source/snowflake.py
new file mode 100644
index 0000000000..4b91563ac7
--- /dev/null
+++ b/metadata-ingestion/src/gometa/ingestion/source/snowflake.py
@@ -0,0 +1,16 @@
+from .sql_common import SQLAlchemyConfig, SQLAlchemySource
+
+
+class SnowflakeConfig(SQLAlchemyConfig):
+    # defaults
+    scheme = "snowflake"
+
+
+class SnowflakeSource(SQLAlchemySource):
+    def __init__(self, config, ctx):
+        super().__init__(config, ctx, "snowflake")
+
+    @classmethod
+    def create(cls, config_dict, ctx):
+        config = SnowflakeConfig.parse_obj(config_dict)
+        return cls(config, ctx)